diff --git a/cmd/initialise/helper.go b/cmd/initialise/helper.go index ac212e8e6..94d5aef7e 100644 --- a/cmd/initialise/helper.go +++ b/cmd/initialise/helper.go @@ -1,6 +1,7 @@ package initialise import ( + "context" "errors" "github.com/jackc/pgx/v5/pgconn" @@ -8,8 +9,8 @@ import ( "github.com/zitadel/zitadel/internal/database" ) -func exec(db *database.DB, stmt string, possibleErrCodes []string, args ...interface{}) error { - _, err := db.Exec(stmt, args...) +func exec(ctx context.Context, db database.ContextExecuter, stmt string, possibleErrCodes []string, args ...interface{}) error { + _, err := db.ExecContext(ctx, stmt, args...) pgErr := new(pgconn.PgError) if errors.As(err, &pgErr) { for _, possibleCode := range possibleErrCodes { diff --git a/cmd/initialise/init.go b/cmd/initialise/init.go index 917e6a2d9..fba5098fa 100644 --- a/cmd/initialise/init.go +++ b/cmd/initialise/init.go @@ -59,7 +59,7 @@ The user provided by flags needs privileges to } func InitAll(ctx context.Context, config *Config) { - err := initialise(config.Database, + err := initialise(ctx, config.Database, VerifyUser(config.Database.Username(), config.Database.Password()), VerifyDatabase(config.Database.DatabaseName()), VerifyGrant(config.Database.DatabaseName(), config.Database.Username()), @@ -71,7 +71,7 @@ func InitAll(ctx context.Context, config *Config) { logging.OnError(err).Fatal("unable to initialize ZITADEL") } -func initialise(config database.Config, steps ...func(*database.DB) error) error { +func initialise(ctx context.Context, config database.Config, steps ...func(context.Context, *database.DB) error) error { logging.Info("initialization started") err := ReadStmts(config.Type()) @@ -85,12 +85,12 @@ func initialise(config database.Config, steps ...func(*database.DB) error) error } defer db.Close() - return Init(db, steps...) + return Init(ctx, db, steps...) } -func Init(db *database.DB, steps ...func(*database.DB) error) error { +func Init(ctx context.Context, db *database.DB, steps ...func(context.Context, *database.DB) error) error { for _, step := range steps { - if err := step(db); err != nil { + if err := step(ctx, db); err != nil { return err } } diff --git a/cmd/initialise/sql/cockroach/02_database.sql b/cmd/initialise/sql/cockroach/02_database.sql index a0e3c3350..6103b95b3 100644 --- a/cmd/initialise/sql/cockroach/02_database.sql +++ b/cmd/initialise/sql/cockroach/02_database.sql @@ -1,2 +1,2 @@ -- replace %[1]s with the name of the database -CREATE DATABASE IF NOT EXISTS "%[1]s" \ No newline at end of file +CREATE DATABASE IF NOT EXISTS "%[1]s"; diff --git a/cmd/initialise/sql/cockroach/08_events_table.sql b/cmd/initialise/sql/cockroach/08_events_table.sql index 00eba689f..ebaf18ce2 100644 --- a/cmd/initialise/sql/cockroach/08_events_table.sql +++ b/cmd/initialise/sql/cockroach/08_events_table.sql @@ -19,3 +19,98 @@ CREATE TABLE IF NOT EXISTS eventstore.events2 ( , INDEX es_wm (aggregate_id, instance_id, aggregate_type, event_type) , INDEX es_projection (instance_id, aggregate_type, event_type, "position" DESC) ); + +-- represents an event to be created. +CREATE TYPE IF NOT EXISTS eventstore.command AS ( + instance_id TEXT + , aggregate_type TEXT + , aggregate_id TEXT + , command_type TEXT + , revision INT2 + , payload JSONB + , creator TEXT + , owner TEXT +); + +CREATE OR REPLACE FUNCTION eventstore.commands_to_events(commands eventstore.command[]) RETURNS SETOF eventstore.events2 VOLATILE AS $$ +SELECT + ("c").instance_id + , ("c").aggregate_type + , ("c").aggregate_id + , ("c").command_type AS event_type + , cs.sequence + ROW_NUMBER() OVER (PARTITION BY ("c").instance_id, ("c").aggregate_type, ("c").aggregate_id ORDER BY ("c").in_tx_order) AS sequence + , ("c").revision + , hlc_to_timestamp(cluster_logical_timestamp()) AS created_at + , ("c").payload + , ("c").creator + , cs.owner + , cluster_logical_timestamp() AS position + , ("c").in_tx_order +FROM ( + SELECT + ("c").instance_id + , ("c").aggregate_type + , ("c").aggregate_id + , ("c").command_type + , ("c").revision + , ("c").payload + , ("c").creator + , ("c").owner + , ROW_NUMBER() OVER () AS in_tx_order + FROM + UNNEST(commands) AS "c" +) AS "c" +JOIN ( + SELECT + cmds.instance_id + , cmds.aggregate_type + , cmds.aggregate_id + , CASE WHEN (e.owner IS NOT NULL OR e.owner <> '') THEN e.owner ELSE command_owners.owner END AS owner + , COALESCE(MAX(e.sequence), 0) AS sequence + FROM ( + SELECT DISTINCT + ("cmds").instance_id + , ("cmds").aggregate_type + , ("cmds").aggregate_id + , ("cmds").owner + FROM UNNEST(commands) AS "cmds" + ) AS cmds + LEFT JOIN eventstore.events2 AS e + ON cmds.instance_id = e.instance_id + AND cmds.aggregate_type = e.aggregate_type + AND cmds.aggregate_id = e.aggregate_id + JOIN ( + SELECT + DISTINCT ON ( + ("c").instance_id + , ("c").aggregate_type + , ("c").aggregate_id + ) + ("c").instance_id + , ("c").aggregate_type + , ("c").aggregate_id + , ("c").owner + FROM + UNNEST(commands) AS "c" + ) AS command_owners ON + cmds.instance_id = command_owners.instance_id + AND cmds.aggregate_type = command_owners.aggregate_type + AND cmds.aggregate_id = command_owners.aggregate_id + GROUP BY + cmds.instance_id + , cmds.aggregate_type + , cmds.aggregate_id + , 4 -- owner +) AS cs + ON ("c").instance_id = cs.instance_id + AND ("c").aggregate_type = cs.aggregate_type + AND ("c").aggregate_id = cs.aggregate_id +ORDER BY + in_tx_order +$$ LANGUAGE SQL; + +CREATE OR REPLACE FUNCTION eventstore.push(commands eventstore.command[]) RETURNS SETOF eventstore.events2 AS $$ + INSERT INTO eventstore.events2 + SELECT * FROM eventstore.commands_to_events(commands) + RETURNING * +$$ LANGUAGE SQL; \ No newline at end of file diff --git a/cmd/initialise/sql/postgres/08_events_table.sql b/cmd/initialise/sql/postgres/08_events_table.sql index d978df5d9..c4b7351e6 100644 --- a/cmd/initialise/sql/postgres/08_events_table.sql +++ b/cmd/initialise/sql/postgres/08_events_table.sql @@ -19,4 +19,103 @@ CREATE TABLE IF NOT EXISTS eventstore.events2 ( CREATE INDEX IF NOT EXISTS es_active_instances ON eventstore.events2 (created_at DESC, instance_id); CREATE INDEX IF NOT EXISTS es_wm ON eventstore.events2 (aggregate_id, instance_id, aggregate_type, event_type); -CREATE INDEX IF NOT EXISTS es_projection ON eventstore.events2 (instance_id, aggregate_type, event_type, "position"); \ No newline at end of file +CREATE INDEX IF NOT EXISTS es_projection ON eventstore.events2 (instance_id, aggregate_type, event_type, "position"); + +-- represents an event to be created. +DO $$ BEGIN + CREATE TYPE eventstore.command AS ( + instance_id TEXT + , aggregate_type TEXT + , aggregate_id TEXT + , command_type TEXT + , revision INT2 + , payload JSONB + , creator TEXT + , owner TEXT + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; + +CREATE OR REPLACE FUNCTION eventstore.commands_to_events(commands eventstore.command[]) RETURNS SETOF eventstore.events2 VOLATILE AS $$ +SELECT + c.instance_id + , c.aggregate_type + , c.aggregate_id + , c.command_type AS event_type + , cs.sequence + ROW_NUMBER() OVER (PARTITION BY c.instance_id, c.aggregate_type, c.aggregate_id ORDER BY c.in_tx_order) AS sequence + , c.revision + , NOW() AS created_at + , c.payload + , c.creator + , cs.owner + , EXTRACT(EPOCH FROM NOW()) AS position + , c.in_tx_order +FROM ( + SELECT + c.instance_id + , c.aggregate_type + , c.aggregate_id + , c.command_type + , c.revision + , c.payload + , c.creator + , c.owner + , ROW_NUMBER() OVER () AS in_tx_order + FROM + UNNEST(commands) AS c +) AS c +JOIN ( + SELECT + cmds.instance_id + , cmds.aggregate_type + , cmds.aggregate_id + , CASE WHEN (e.owner IS NOT NULL OR e.owner <> '') THEN e.owner ELSE command_owners.owner END AS owner + , COALESCE(MAX(e.sequence), 0) AS sequence + FROM ( + SELECT DISTINCT + instance_id + , aggregate_type + , aggregate_id + , owner + FROM UNNEST(commands) + ) AS cmds + LEFT JOIN eventstore.events2 AS e + ON cmds.instance_id = e.instance_id + AND cmds.aggregate_type = e.aggregate_type + AND cmds.aggregate_id = e.aggregate_id + JOIN ( + SELECT + DISTINCT ON ( + instance_id + , aggregate_type + , aggregate_id + ) + instance_id + , aggregate_type + , aggregate_id + , owner + FROM + UNNEST(commands) + ) AS command_owners ON + cmds.instance_id = command_owners.instance_id + AND cmds.aggregate_type = command_owners.aggregate_type + AND cmds.aggregate_id = command_owners.aggregate_id + GROUP BY + cmds.instance_id + , cmds.aggregate_type + , cmds.aggregate_id + , 4 -- owner +) AS cs + ON c.instance_id = cs.instance_id + AND c.aggregate_type = cs.aggregate_type + AND c.aggregate_id = cs.aggregate_id +ORDER BY + in_tx_order; +$$ LANGUAGE SQL; + +CREATE OR REPLACE FUNCTION eventstore.push(commands eventstore.command[]) RETURNS SETOF eventstore.events2 VOLATILE AS $$ +INSERT INTO eventstore.events2 +SELECT * FROM eventstore.commands_to_events(commands) +RETURNING * +$$ LANGUAGE SQL; diff --git a/cmd/initialise/verify_database.go b/cmd/initialise/verify_database.go index be3ec19ba..6e04e489f 100644 --- a/cmd/initialise/verify_database.go +++ b/cmd/initialise/verify_database.go @@ -1,6 +1,7 @@ package initialise import ( + "context" _ "embed" "fmt" @@ -28,16 +29,16 @@ The user provided by flags needs privileges to Run: func(cmd *cobra.Command, args []string) { config := MustNewConfig(viper.GetViper()) - err := initialise(config.Database, VerifyDatabase(config.Database.DatabaseName())) + err := initialise(cmd.Context(), config.Database, VerifyDatabase(config.Database.DatabaseName())) logging.OnError(err).Fatal("unable to initialize the database") }, } } -func VerifyDatabase(databaseName string) func(*database.DB) error { - return func(db *database.DB) error { +func VerifyDatabase(databaseName string) func(context.Context, *database.DB) error { + return func(ctx context.Context, db *database.DB) error { logging.WithFields("database", databaseName).Info("verify database") - return exec(db, fmt.Sprintf(databaseStmt, databaseName), []string{dbAlreadyExistsCode}) + return exec(ctx, db, fmt.Sprintf(databaseStmt, databaseName), []string{dbAlreadyExistsCode}) } } diff --git a/cmd/initialise/verify_database_test.go b/cmd/initialise/verify_database_test.go index ebdf0473b..d7da97847 100644 --- a/cmd/initialise/verify_database_test.go +++ b/cmd/initialise/verify_database_test.go @@ -1,6 +1,7 @@ package initialise import ( + "context" "database/sql" "errors" "testing" @@ -55,7 +56,7 @@ func Test_verifyDB(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if err := VerifyDatabase(tt.args.database)(tt.args.db.db); !errors.Is(err, tt.targetErr) { + if err := VerifyDatabase(tt.args.database)(context.Background(), tt.args.db.db); !errors.Is(err, tt.targetErr) { t.Errorf("verifyDB() error = %v, want: %v", err, tt.targetErr) } if err := tt.args.db.mock.ExpectationsWereMet(); err != nil { diff --git a/cmd/initialise/verify_grant.go b/cmd/initialise/verify_grant.go index ed8ecb725..a14a495bf 100644 --- a/cmd/initialise/verify_grant.go +++ b/cmd/initialise/verify_grant.go @@ -1,6 +1,7 @@ package initialise import ( + "context" _ "embed" "fmt" @@ -23,16 +24,16 @@ Prerequisites: Run: func(cmd *cobra.Command, args []string) { config := MustNewConfig(viper.GetViper()) - err := initialise(config.Database, VerifyGrant(config.Database.DatabaseName(), config.Database.Username())) + err := initialise(cmd.Context(), config.Database, VerifyGrant(config.Database.DatabaseName(), config.Database.Username())) logging.OnError(err).Fatal("unable to set grant") }, } } -func VerifyGrant(databaseName, username string) func(*database.DB) error { - return func(db *database.DB) error { +func VerifyGrant(databaseName, username string) func(context.Context, *database.DB) error { + return func(ctx context.Context, db *database.DB) error { logging.WithFields("user", username, "database", databaseName).Info("verify grant") - return exec(db, fmt.Sprintf(grantStmt, databaseName, username), nil) + return exec(ctx, db, fmt.Sprintf(grantStmt, databaseName, username), nil) } } diff --git a/cmd/initialise/verify_grant_test.go b/cmd/initialise/verify_grant_test.go index a6bfa818a..3cab9099f 100644 --- a/cmd/initialise/verify_grant_test.go +++ b/cmd/initialise/verify_grant_test.go @@ -1,6 +1,7 @@ package initialise import ( + "context" "database/sql" "errors" "testing" @@ -53,7 +54,7 @@ func Test_verifyGrant(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if err := VerifyGrant(tt.args.database, tt.args.username)(tt.args.db.db); !errors.Is(err, tt.targetErr) { + if err := VerifyGrant(tt.args.database, tt.args.username)(context.Background(), tt.args.db.db); !errors.Is(err, tt.targetErr) { t.Errorf("VerifyGrant() error = %v, want: %v", err, tt.targetErr) } if err := tt.args.db.mock.ExpectationsWereMet(); err != nil { diff --git a/cmd/initialise/verify_settings.go b/cmd/initialise/verify_settings.go index 75e811663..6f4ba7c07 100644 --- a/cmd/initialise/verify_settings.go +++ b/cmd/initialise/verify_settings.go @@ -1,6 +1,7 @@ package initialise import ( + "context" _ "embed" "fmt" @@ -26,19 +27,19 @@ Cockroach Run: func(cmd *cobra.Command, args []string) { config := MustNewConfig(viper.GetViper()) - err := initialise(config.Database, VerifySettings(config.Database.DatabaseName(), config.Database.Username())) + err := initialise(cmd.Context(), config.Database, VerifySettings(config.Database.DatabaseName(), config.Database.Username())) logging.OnError(err).Fatal("unable to set settings") }, } } -func VerifySettings(databaseName, username string) func(*database.DB) error { - return func(db *database.DB) error { +func VerifySettings(databaseName, username string) func(context.Context, *database.DB) error { + return func(ctx context.Context, db *database.DB) error { if db.Type() == "postgres" { return nil } logging.WithFields("user", username, "database", databaseName).Info("verify settings") - return exec(db, fmt.Sprintf(settingsStmt, databaseName, username), nil) + return exec(ctx, db, fmt.Sprintf(settingsStmt, databaseName, username), nil) } } diff --git a/cmd/initialise/verify_user.go b/cmd/initialise/verify_user.go index 86e6c25ba..43bdb9142 100644 --- a/cmd/initialise/verify_user.go +++ b/cmd/initialise/verify_user.go @@ -1,6 +1,7 @@ package initialise import ( + "context" _ "embed" "fmt" @@ -28,20 +29,20 @@ The user provided by flags needs privileges to Run: func(cmd *cobra.Command, args []string) { config := MustNewConfig(viper.GetViper()) - err := initialise(config.Database, VerifyUser(config.Database.Username(), config.Database.Password())) + err := initialise(cmd.Context(), config.Database, VerifyUser(config.Database.Username(), config.Database.Password())) logging.OnError(err).Fatal("unable to init user") }, } } -func VerifyUser(username, password string) func(*database.DB) error { - return func(db *database.DB) error { +func VerifyUser(username, password string) func(context.Context, *database.DB) error { + return func(ctx context.Context, db *database.DB) error { logging.WithFields("username", username).Info("verify user") if password != "" { createUserStmt += " WITH PASSWORD '" + password + "'" } - return exec(db, fmt.Sprintf(createUserStmt, username), []string{roleAlreadyExistsCode}) + return exec(ctx, db, fmt.Sprintf(createUserStmt, username), []string{roleAlreadyExistsCode}) } } diff --git a/cmd/initialise/verify_user_test.go b/cmd/initialise/verify_user_test.go index da7afc176..53b35e67d 100644 --- a/cmd/initialise/verify_user_test.go +++ b/cmd/initialise/verify_user_test.go @@ -1,6 +1,7 @@ package initialise import ( + "context" "database/sql" "errors" "testing" @@ -70,7 +71,7 @@ func Test_verifyUser(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if err := VerifyUser(tt.args.username, tt.args.password)(tt.args.db.db); !errors.Is(err, tt.targetErr) { + if err := VerifyUser(tt.args.username, tt.args.password)(context.Background(), tt.args.db.db); !errors.Is(err, tt.targetErr) { t.Errorf("VerifyGrant() error = %v, want: %v", err, tt.targetErr) } if err := tt.args.db.mock.ExpectationsWereMet(); err != nil { diff --git a/cmd/initialise/verify_zitadel.go b/cmd/initialise/verify_zitadel.go index 7c16fdaad..a5ce1fd57 100644 --- a/cmd/initialise/verify_zitadel.go +++ b/cmd/initialise/verify_zitadel.go @@ -2,6 +2,7 @@ package initialise import ( "context" + "database/sql" _ "embed" "fmt" @@ -11,6 +12,7 @@ import ( "github.com/zitadel/zitadel/internal/database" "github.com/zitadel/zitadel/internal/database/dialect" + es_v3 "github.com/zitadel/zitadel/internal/eventstore/v3" ) func newZitadel() *cobra.Command { @@ -36,38 +38,44 @@ func VerifyZitadel(ctx context.Context, db *database.DB, config database.Config) return err } + conn, err := db.Conn(ctx) + if err != nil { + return err + } + defer conn.Close() + logging.WithFields().Info("verify system") - if err := exec(db, fmt.Sprintf(createSystemStmt, config.Username()), nil); err != nil { + if err := exec(ctx, conn, fmt.Sprintf(createSystemStmt, config.Username()), nil); err != nil { return err } logging.WithFields().Info("verify encryption keys") - if err := createEncryptionKeys(ctx, db); err != nil { + if err := createEncryptionKeys(ctx, conn); err != nil { return err } logging.WithFields().Info("verify projections") - if err := exec(db, fmt.Sprintf(createProjectionsStmt, config.Username()), nil); err != nil { + if err := exec(ctx, conn, fmt.Sprintf(createProjectionsStmt, config.Username()), nil); err != nil { return err } logging.WithFields().Info("verify eventstore") - if err := exec(db, fmt.Sprintf(createEventstoreStmt, config.Username()), nil); err != nil { + if err := exec(ctx, conn, fmt.Sprintf(createEventstoreStmt, config.Username()), nil); err != nil { return err } logging.WithFields().Info("verify events tables") - if err := createEvents(ctx, db); err != nil { + if err := createEvents(ctx, conn); err != nil { return err } logging.WithFields().Info("verify system sequence") - if err := exec(db, createSystemSequenceStmt, nil); err != nil { + if err := exec(ctx, conn, createSystemSequenceStmt, nil); err != nil { return err } logging.WithFields().Info("verify unique constraints") - if err := exec(db, createUniqueConstraints, nil); err != nil { + if err := exec(ctx, conn, createUniqueConstraints, nil); err != nil { return err } @@ -89,7 +97,7 @@ func verifyZitadel(ctx context.Context, config database.Config) error { return db.Close() } -func createEncryptionKeys(ctx context.Context, db *database.DB) error { +func createEncryptionKeys(ctx context.Context, db database.Beginner) error { tx, err := db.BeginTx(ctx, nil) if err != nil { return err @@ -103,8 +111,8 @@ func createEncryptionKeys(ctx context.Context, db *database.DB) error { return tx.Commit() } -func createEvents(ctx context.Context, db *database.DB) (err error) { - tx, err := db.BeginTx(ctx, nil) +func createEvents(ctx context.Context, conn *sql.Conn) (err error) { + tx, err := conn.BeginTx(ctx, nil) if err != nil { return err } @@ -127,5 +135,8 @@ func createEvents(ctx context.Context, db *database.DB) (err error) { return row.Err() } _, err = tx.Exec(createEventsStmt) - return err + if err != nil { + return err + } + return es_v3.CheckExecutionPlan(ctx, conn) } diff --git a/cmd/initialise/verify_zitadel_test.go b/cmd/initialise/verify_zitadel_test.go index 64df01bdb..194911a17 100644 --- a/cmd/initialise/verify_zitadel_test.go +++ b/cmd/initialise/verify_zitadel_test.go @@ -108,7 +108,12 @@ func Test_verifyEvents(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if err := createEvents(context.Background(), tt.args.db.db); !errors.Is(err, tt.targetErr) { + conn, err := tt.args.db.db.Conn(context.Background()) + if err != nil { + t.Error(err) + return + } + if err := createEvents(context.Background(), conn); !errors.Is(err, tt.targetErr) { t.Errorf("createEvents() error = %v, want: %v", err, tt.targetErr) } if err := tt.args.db.mock.ExpectationsWereMet(); err != nil { diff --git a/cmd/setup/40.go b/cmd/setup/40.go new file mode 100644 index 000000000..a0d1afcf5 --- /dev/null +++ b/cmd/setup/40.go @@ -0,0 +1,52 @@ +package setup + +import ( + "context" + "embed" + "fmt" + + "github.com/zitadel/logging" + + "github.com/zitadel/zitadel/internal/database" + "github.com/zitadel/zitadel/internal/eventstore" +) + +var ( + //go:embed 40/cockroach/*.sql + //go:embed 40/postgres/*.sql + initPushFunc embed.FS +) + +type InitPushFunc struct { + dbClient *database.DB +} + +func (mig *InitPushFunc) Execute(ctx context.Context, _ eventstore.Event) (err error) { + statements, err := readStatements(initPushFunc, "40", mig.dbClient.Type()) + if err != nil { + return err + } + conn, err := mig.dbClient.Conn(ctx) + if err != nil { + return err + } + defer func() { + closeErr := conn.Close() + logging.OnError(closeErr).Debug("failed to release connection") + // Force the pool to reopen connections to apply the new types + mig.dbClient.Pool.Reset() + }() + + for _, stmt := range statements { + logging.WithFields("file", stmt.file, "migration", mig.String()).Info("execute statement") + if _, err := conn.ExecContext(ctx, stmt.query); err != nil { + return fmt.Errorf("%s %s: %w", mig.String(), stmt.file, err) + } + } + + return nil +} + +func (mig *InitPushFunc) String() string { + return "40_init_push_func" +} diff --git a/cmd/setup/40/cockroach/40_init_push_func.sql b/cmd/setup/40/cockroach/40_init_push_func.sql new file mode 100644 index 000000000..c2e2e92b0 --- /dev/null +++ b/cmd/setup/40/cockroach/40_init_push_func.sql @@ -0,0 +1,107 @@ +-- represents an event to be created. +CREATE TYPE IF NOT EXISTS eventstore.command AS ( + instance_id TEXT + , aggregate_type TEXT + , aggregate_id TEXT + , command_type TEXT + , revision INT2 + , payload JSONB + , creator TEXT + , owner TEXT +); + +/* +select * from eventstore.commands_to_events( +ARRAY[ + ROW('', 'system', 'SYSTEM', 'ct1', 1, '{"key": "value"}', 'c1', 'SYSTEM') + , ROW('', 'system', 'SYSTEM', 'ct2', 1, '{"key": "value"}', 'c1', 'SYSTEM') + , ROW('289525561255060732', 'org', '289575074711790844', 'ct3', 1, '{"key": "value"}', 'c1', '289575074711790844') + , ROW('289525561255060732', 'user', '289575075164906748', 'ct3', 1, '{"key": "value"}', 'c1', '289575074711790844') + , ROW('289525561255060732', 'oidc_session', 'V2_289575178579535100', 'ct3', 1, '{"key": "value"}', 'c1', '289575074711790844') + , ROW('', 'system', 'SYSTEM', 'ct3', 1, '{"key": "value"}', 'c1', 'SYSTEM') +]::eventstore.command[] +); +*/ + +CREATE OR REPLACE FUNCTION eventstore.commands_to_events(commands eventstore.command[]) RETURNS SETOF eventstore.events2 VOLATILE AS $$ +SELECT + ("c").instance_id + , ("c").aggregate_type + , ("c").aggregate_id + , ("c").command_type AS event_type + , cs.sequence + ROW_NUMBER() OVER (PARTITION BY ("c").instance_id, ("c").aggregate_type, ("c").aggregate_id ORDER BY ("c").in_tx_order) AS sequence + , ("c").revision + , hlc_to_timestamp(cluster_logical_timestamp()) AS created_at + , ("c").payload + , ("c").creator + , cs.owner + , cluster_logical_timestamp() AS position + , ("c").in_tx_order +FROM ( + SELECT + ("c").instance_id + , ("c").aggregate_type + , ("c").aggregate_id + , ("c").command_type + , ("c").revision + , ("c").payload + , ("c").creator + , ("c").owner + , ROW_NUMBER() OVER () AS in_tx_order + FROM + UNNEST(commands) AS "c" +) AS "c" +JOIN ( + SELECT + cmds.instance_id + , cmds.aggregate_type + , cmds.aggregate_id + , CASE WHEN (e.owner <> '') THEN e.owner ELSE command_owners.owner END AS owner + , COALESCE(MAX(e.sequence), 0) AS sequence + FROM ( + SELECT DISTINCT + ("cmds").instance_id + , ("cmds").aggregate_type + , ("cmds").aggregate_id + , ("cmds").owner + FROM UNNEST(commands) AS "cmds" + ) AS cmds + LEFT JOIN eventstore.events2 AS e + ON cmds.instance_id = e.instance_id + AND cmds.aggregate_type = e.aggregate_type + AND cmds.aggregate_id = e.aggregate_id + JOIN ( + SELECT + DISTINCT ON ( + ("c").instance_id + , ("c").aggregate_type + , ("c").aggregate_id + ) + ("c").instance_id + , ("c").aggregate_type + , ("c").aggregate_id + , ("c").owner + FROM + UNNEST(commands) AS "c" + ) AS command_owners ON + cmds.instance_id = command_owners.instance_id + AND cmds.aggregate_type = command_owners.aggregate_type + AND cmds.aggregate_id = command_owners.aggregate_id + GROUP BY + cmds.instance_id + , cmds.aggregate_type + , cmds.aggregate_id + , 4 -- owner +) AS cs + ON ("c").instance_id = cs.instance_id + AND ("c").aggregate_type = cs.aggregate_type + AND ("c").aggregate_id = cs.aggregate_id +ORDER BY + in_tx_order +$$ LANGUAGE SQL; + +CREATE OR REPLACE FUNCTION eventstore.push(commands eventstore.command[]) RETURNS SETOF eventstore.events2 AS $$ + INSERT INTO eventstore.events2 + SELECT * FROM eventstore.commands_to_events(commands) + RETURNING * +$$ LANGUAGE SQL; \ No newline at end of file diff --git a/cmd/setup/40/postgres/01_type.sql b/cmd/setup/40/postgres/01_type.sql new file mode 100644 index 000000000..ace6d8fe1 --- /dev/null +++ b/cmd/setup/40/postgres/01_type.sql @@ -0,0 +1,15 @@ +-- represents an event to be created. +DO $$ BEGIN + CREATE TYPE eventstore.command AS ( + instance_id TEXT + , aggregate_type TEXT + , aggregate_id TEXT + , command_type TEXT + , revision INT2 + , payload JSONB + , creator TEXT + , owner TEXT + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; diff --git a/cmd/setup/40/postgres/02_func.sql b/cmd/setup/40/postgres/02_func.sql new file mode 100644 index 000000000..5f84f3908 --- /dev/null +++ b/cmd/setup/40/postgres/02_func.sql @@ -0,0 +1,82 @@ +CREATE OR REPLACE FUNCTION eventstore.commands_to_events(commands eventstore.command[]) RETURNS SETOF eventstore.events2 VOLATILE AS $$ +SELECT + c.instance_id + , c.aggregate_type + , c.aggregate_id + , c.command_type AS event_type + , cs.sequence + ROW_NUMBER() OVER (PARTITION BY c.instance_id, c.aggregate_type, c.aggregate_id ORDER BY c.in_tx_order) AS sequence + , c.revision + , NOW() AS created_at + , c.payload + , c.creator + , cs.owner + , EXTRACT(EPOCH FROM NOW()) AS position + , c.in_tx_order +FROM ( + SELECT + c.instance_id + , c.aggregate_type + , c.aggregate_id + , c.command_type + , c.revision + , c.payload + , c.creator + , c.owner + , ROW_NUMBER() OVER () AS in_tx_order + FROM + UNNEST(commands) AS c +) AS c +JOIN ( + SELECT + cmds.instance_id + , cmds.aggregate_type + , cmds.aggregate_id + , CASE WHEN (e.owner IS NOT NULL OR e.owner <> '') THEN e.owner ELSE command_owners.owner END AS owner + , COALESCE(MAX(e.sequence), 0) AS sequence + FROM ( + SELECT DISTINCT + instance_id + , aggregate_type + , aggregate_id + , owner + FROM UNNEST(commands) + ) AS cmds + LEFT JOIN eventstore.events2 AS e + ON cmds.instance_id = e.instance_id + AND cmds.aggregate_type = e.aggregate_type + AND cmds.aggregate_id = e.aggregate_id + JOIN ( + SELECT + DISTINCT ON ( + instance_id + , aggregate_type + , aggregate_id + ) + instance_id + , aggregate_type + , aggregate_id + , owner + FROM + UNNEST(commands) + ) AS command_owners ON + cmds.instance_id = command_owners.instance_id + AND cmds.aggregate_type = command_owners.aggregate_type + AND cmds.aggregate_id = command_owners.aggregate_id + GROUP BY + cmds.instance_id + , cmds.aggregate_type + , cmds.aggregate_id + , 4 -- owner +) AS cs + ON c.instance_id = cs.instance_id + AND c.aggregate_type = cs.aggregate_type + AND c.aggregate_id = cs.aggregate_id +ORDER BY + in_tx_order; +$$ LANGUAGE SQL; + +CREATE OR REPLACE FUNCTION eventstore.push(commands eventstore.command[]) RETURNS SETOF eventstore.events2 VOLATILE AS $$ +INSERT INTO eventstore.events2 +SELECT * FROM eventstore.commands_to_events(commands) +RETURNING * +$$ LANGUAGE SQL; diff --git a/cmd/setup/config.go b/cmd/setup/config.go index bd5444f2d..34bc80d4a 100644 --- a/cmd/setup/config.go +++ b/cmd/setup/config.go @@ -126,6 +126,7 @@ type Steps struct { s36FillV2Milestones *FillV3Milestones s37Apps7OIDConfigsBackChannelLogoutURI *Apps7OIDConfigsBackChannelLogoutURI s38BackChannelLogoutNotificationStart *BackChannelLogoutNotificationStart + s40InitPushFunc *InitPushFunc s39DeleteStaleOrgFields *DeleteStaleOrgFields } diff --git a/cmd/setup/setup.go b/cmd/setup/setup.go index e9721c6b3..8322e081e 100644 --- a/cmd/setup/setup.go +++ b/cmd/setup/setup.go @@ -170,6 +170,7 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string) 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) logging.OnError(err).Fatal("unable to start projections") @@ -190,6 +191,7 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string) for _, step := range []migration.Migration{ steps.s14NewEventsTable, + steps.s40InitPushFunc, steps.s1ProjectionTable, steps.s2AssetsTable, steps.s28AddFieldTable, diff --git a/internal/api/grpc/resources/webkey/v3/integration_test/webkey_integration_test.go b/internal/api/grpc/resources/webkey/v3/integration_test/webkey_integration_test.go index 19d02dcea..eafa733fd 100644 --- a/internal/api/grpc/resources/webkey/v3/integration_test/webkey_integration_test.go +++ b/internal/api/grpc/resources/webkey/v3/integration_test/webkey_integration_test.go @@ -216,6 +216,7 @@ func assertFeatureDisabledError(t *testing.T, err error) { } func checkWebKeyListState(ctx context.Context, t *testing.T, instance *integration.Instance, nKeys int, expectActiveKeyID string, config any, creationDate *timestamppb.Timestamp) { + t.Helper() retryDuration, tick := integration.WaitForAndTickWithMaxDuration(ctx, time.Minute) assert.EventuallyWithT(t, func(collect *assert.CollectT) { diff --git a/internal/database/cockroach/crdb.go b/internal/database/cockroach/crdb.go index 527becf7b..cc89be868 100644 --- a/internal/database/cockroach/crdb.go +++ b/internal/database/cockroach/crdb.go @@ -3,15 +3,18 @@ package cockroach import ( "context" "database/sql" + "fmt" "strconv" "strings" "time" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "github.com/jackc/pgx/v5/stdlib" "github.com/mitchellh/mapstructure" "github.com/zitadel/logging" + "github.com/zitadel/zitadel/internal/api/authz" "github.com/zitadel/zitadel/internal/database/dialect" ) @@ -72,6 +75,12 @@ func (_ *Config) Decode(configs []interface{}) (dialect.Connector, error) { } func (c *Config) Connect(useAdmin bool, pusherRatio, spoolerRatio float64, purpose dialect.DBPurpose) (*sql.DB, *pgxpool.Pool, error) { + dialect.RegisterAfterConnect(func(ctx context.Context, c *pgx.Conn) error { + // CockroachDB by default does not allow multiple modifications of the same table using ON CONFLICT + // This is needed to fill the fields table of the eventstore during eventstore.Push. + _, err := c.Exec(ctx, "SET enable_multiple_modifications_of_table = on") + return err + }) connConfig, err := dialect.NewConnectionConfig(c.MaxOpenConns, c.MaxIdleConns, pusherRatio, spoolerRatio, purpose) if err != nil { return nil, nil, err @@ -82,6 +91,29 @@ func (c *Config) Connect(useAdmin bool, pusherRatio, spoolerRatio float64, purpo return nil, nil, err } + if len(connConfig.AfterConnect) > 0 { + config.AfterConnect = func(ctx context.Context, conn *pgx.Conn) error { + for _, f := range connConfig.AfterConnect { + if err := f(ctx, conn); err != nil { + return err + } + } + return nil + } + } + + // For the pusher we set the app name with the instance ID + if purpose == dialect.DBPurposeEventPusher { + config.BeforeAcquire = func(ctx context.Context, conn *pgx.Conn) bool { + return setAppNameWithID(ctx, conn, purpose, authz.GetInstance(ctx).InstanceID()) + } + config.AfterRelease = func(conn *pgx.Conn) bool { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + return setAppNameWithID(ctx, conn, purpose, "IDLE") + } + } + if connConfig.MaxOpenConns != 0 { config.MaxConns = int32(connConfig.MaxOpenConns) } @@ -200,3 +232,11 @@ func (c Config) String(useAdmin bool, appName string) string { return strings.Join(fields, " ") } + +func setAppNameWithID(ctx context.Context, conn *pgx.Conn, purpose dialect.DBPurpose, id string) bool { + // needs to be set like this because psql complains about parameters in the SET statement + query := fmt.Sprintf("SET application_name = '%s_%s'", purpose.AppName(), id) + _, err := conn.Exec(ctx, query) + logging.OnError(err).Warn("failed to set application name") + return err == nil +} diff --git a/internal/database/database.go b/internal/database/database.go index 0191f34b6..b86a9f247 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -18,21 +18,31 @@ import ( "github.com/zitadel/zitadel/internal/zerrors" ) -type QueryExecuter interface { - Query(query string, args ...any) (*sql.Rows, error) +type ContextQuerier interface { QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) - Exec(query string, args ...any) (sql.Result, error) +} + +type ContextExecuter interface { ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) } +type ContextQueryExecuter interface { + ContextQuerier + ContextExecuter +} + type Client interface { - QueryExecuter + ContextQueryExecuter + Beginner + Conn(ctx context.Context) (*sql.Conn, error) +} + +type Beginner interface { BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error) - Begin() (*sql.Tx, error) } type Tx interface { - QueryExecuter + ContextQueryExecuter Commit() error Rollback() error } diff --git a/internal/database/dialect/connections.go b/internal/database/dialect/connections.go index 48f8d6e22..f957870df 100644 --- a/internal/database/dialect/connections.go +++ b/internal/database/dialect/connections.go @@ -1,8 +1,13 @@ package dialect import ( + "context" "errors" "fmt" + "reflect" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" ) var ( @@ -17,6 +22,7 @@ var ( type ConnectionConfig struct { MaxOpenConns, MaxIdleConns uint32 + AfterConnect []func(ctx context.Context, c *pgx.Conn) error } // takeRatio of MaxOpenConns and MaxIdleConns from config and returns @@ -29,6 +35,7 @@ func (c *ConnectionConfig) takeRatio(ratio float64) (*ConnectionConfig, error) { out := &ConnectionConfig{ MaxOpenConns: uint32(ratio * float64(c.MaxOpenConns)), MaxIdleConns: uint32(ratio * float64(c.MaxIdleConns)), + AfterConnect: c.AfterConnect, } if c.MaxOpenConns != 0 && out.MaxOpenConns < 1 && ratio > 0 { out.MaxOpenConns = 1 @@ -40,6 +47,36 @@ func (c *ConnectionConfig) takeRatio(ratio float64) (*ConnectionConfig, error) { return out, nil } +var afterConnectFuncs []func(ctx context.Context, c *pgx.Conn) error + +func RegisterAfterConnect(f func(ctx context.Context, c *pgx.Conn) error) { + afterConnectFuncs = append(afterConnectFuncs, f) +} + +func RegisterDefaultPgTypeVariants[T any](m *pgtype.Map, name, arrayName string) { + // T + var value T + m.RegisterDefaultPgType(value, name) + + // *T + valueType := reflect.TypeOf(value) + m.RegisterDefaultPgType(reflect.New(valueType).Interface(), name) + + // []T + sliceType := reflect.SliceOf(valueType) + m.RegisterDefaultPgType(reflect.MakeSlice(sliceType, 0, 0).Interface(), arrayName) + + // *[]T + m.RegisterDefaultPgType(reflect.New(sliceType).Interface(), arrayName) + + // []*T + sliceOfPointerType := reflect.SliceOf(reflect.TypeOf(reflect.New(valueType).Interface())) + m.RegisterDefaultPgType(reflect.MakeSlice(sliceOfPointerType, 0, 0).Interface(), arrayName) + + // *[]*T + m.RegisterDefaultPgType(reflect.New(sliceOfPointerType).Interface(), arrayName) +} + // NewConnectionConfig calculates [ConnectionConfig] values from the passed ratios // and returns the config applicable for the requested purpose. // @@ -59,11 +96,13 @@ func NewConnectionConfig(openConns, idleConns uint32, pusherRatio, projectionRat queryConfig := &ConnectionConfig{ MaxOpenConns: openConns, MaxIdleConns: idleConns, + AfterConnect: afterConnectFuncs, } pusherConfig, err := queryConfig.takeRatio(pusherRatio) if err != nil { return nil, fmt.Errorf("event pusher: %w", err) } + spoolerConfig, err := queryConfig.takeRatio(projectionRatio) if err != nil { return nil, fmt.Errorf("projection spooler: %w", err) diff --git a/internal/database/postgres/pg.go b/internal/database/postgres/pg.go index 539aedf0a..c12e12243 100644 --- a/internal/database/postgres/pg.go +++ b/internal/database/postgres/pg.go @@ -3,15 +3,18 @@ package postgres import ( "context" "database/sql" + "fmt" "strconv" "strings" "time" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "github.com/jackc/pgx/v5/stdlib" "github.com/mitchellh/mapstructure" "github.com/zitadel/logging" + "github.com/zitadel/zitadel/internal/api/authz" "github.com/zitadel/zitadel/internal/database/dialect" ) @@ -83,6 +86,27 @@ func (c *Config) Connect(useAdmin bool, pusherRatio, spoolerRatio float64, purpo return nil, nil, err } + config.AfterConnect = func(ctx context.Context, conn *pgx.Conn) error { + for _, f := range connConfig.AfterConnect { + if err := f(ctx, conn); err != nil { + return err + } + } + return nil + } + + // For the pusher we set the app name with the instance ID + if purpose == dialect.DBPurposeEventPusher { + config.BeforeAcquire = func(ctx context.Context, conn *pgx.Conn) bool { + return setAppNameWithID(ctx, conn, purpose, authz.GetInstance(ctx).InstanceID()) + } + config.AfterRelease = func(conn *pgx.Conn) bool { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + return setAppNameWithID(ctx, conn, purpose, "IDLE") + } + } + if connConfig.MaxOpenConns != 0 { config.MaxConns = int32(connConfig.MaxOpenConns) } @@ -209,3 +233,11 @@ func (c Config) String(useAdmin bool, appName string) string { return strings.Join(fields, " ") } + +func setAppNameWithID(ctx context.Context, conn *pgx.Conn, purpose dialect.DBPurpose, id string) bool { + // needs to be set like this because psql complains about parameters in the SET statement + query := fmt.Sprintf("SET application_name = '%s_%s'", purpose.AppName(), id) + _, err := conn.Exec(ctx, query) + logging.OnError(err).Warn("failed to set application name") + return err == nil +} diff --git a/internal/eventstore/event_base.go b/internal/eventstore/event_base.go index c2b56128a..45706641d 100644 --- a/internal/eventstore/event_base.go +++ b/internal/eventstore/event_base.go @@ -3,8 +3,12 @@ package eventstore import ( "context" "encoding/json" + "strconv" + "strings" "time" + "github.com/zitadel/logging" + "github.com/zitadel/zitadel/internal/api/authz" "github.com/zitadel/zitadel/internal/api/service" ) @@ -84,8 +88,10 @@ func (e *BaseEvent) DataAsBytes() []byte { } // Revision implements action -func (*BaseEvent) Revision() uint16 { - return 0 +func (e *BaseEvent) Revision() uint16 { + revision, err := strconv.ParseUint(strings.TrimPrefix(string(e.Agg.Version), "v"), 10, 16) + logging.OnError(err).Debug("failed to parse event revision") + return uint16(revision) } // Unmarshal implements Event diff --git a/internal/eventstore/eventstore.go b/internal/eventstore/eventstore.go index 22dfde3f4..4f331c185 100644 --- a/internal/eventstore/eventstore.go +++ b/internal/eventstore/eventstore.go @@ -90,7 +90,7 @@ func (es *Eventstore) Push(ctx context.Context, cmds ...Command) ([]Event, error // PushWithClient pushes the events in a single transaction using the provided database client // an event needs at least an aggregate -func (es *Eventstore) PushWithClient(ctx context.Context, client database.QueryExecuter, cmds ...Command) ([]Event, error) { +func (es *Eventstore) PushWithClient(ctx context.Context, client database.ContextQueryExecuter, cmds ...Command) ([]Event, error) { if es.PushTimeout > 0 { var cancel func() ctx, cancel = context.WithTimeout(ctx, es.PushTimeout) @@ -301,7 +301,7 @@ type Pusher interface { // Health checks if the connection to the storage is available Health(ctx context.Context) error // Push stores the actions - Push(ctx context.Context, client database.QueryExecuter, commands ...Command) (_ []Event, err error) + Push(ctx context.Context, client database.ContextQueryExecuter, commands ...Command) (_ []Event, err error) // Client returns the underlying database connection Client() *database.DB } diff --git a/internal/eventstore/eventstore_test.go b/internal/eventstore/eventstore_test.go index ec28b9c55..9e1aa77db 100644 --- a/internal/eventstore/eventstore_test.go +++ b/internal/eventstore/eventstore_test.go @@ -347,7 +347,7 @@ func (repo *testPusher) Health(ctx context.Context) error { return nil } -func (repo *testPusher) Push(_ context.Context, _ database.QueryExecuter, commands ...Command) (events []Event, err error) { +func (repo *testPusher) Push(_ context.Context, _ database.ContextQueryExecuter, commands ...Command) (events []Event, err error) { if len(repo.errs) != 0 { err, repo.errs = repo.errs[0], repo.errs[1:] return nil, err @@ -490,6 +490,7 @@ func TestEventstore_Push(t *testing.T) { Type: "test.aggregate", ResourceOwner: "caos", InstanceID: "zitadel", + Version: "v1", }, Data: []byte(nil), User: "editorUser", @@ -534,6 +535,7 @@ func TestEventstore_Push(t *testing.T) { Type: "test.aggregate", ResourceOwner: "caos", InstanceID: "zitadel", + Version: "v1", }, Data: []byte(nil), User: "editorUser", @@ -585,6 +587,7 @@ func TestEventstore_Push(t *testing.T) { Type: "test.aggregate", ResourceOwner: "caos", InstanceID: "zitadel", + Version: "v1", }, Data: []byte(nil), User: "editorUser", @@ -596,6 +599,7 @@ func TestEventstore_Push(t *testing.T) { Type: "test.aggregate", ResourceOwner: "caos", InstanceID: "zitadel", + Version: "v1", }, Data: []byte(nil), User: "editorUser", @@ -658,6 +662,7 @@ func TestEventstore_Push(t *testing.T) { Type: "test.aggregate", ResourceOwner: "caos", InstanceID: "zitadel", + Version: "v1", }, Data: []byte(nil), User: "editorUser", @@ -669,6 +674,7 @@ func TestEventstore_Push(t *testing.T) { Type: "test.aggregate", ResourceOwner: "caos", InstanceID: "zitadel", + Version: "v1", }, Data: []byte(nil), User: "editorUser", @@ -682,6 +688,7 @@ func TestEventstore_Push(t *testing.T) { Type: "test.aggregate", ResourceOwner: "caos", InstanceID: "zitadel", + Version: "v1", }, Data: []byte(nil), User: "editorUser", @@ -778,6 +785,7 @@ func TestEventstore_Push(t *testing.T) { Type: "test.aggregate", ResourceOwner: "caos", InstanceID: "zitadel", + Version: "v1", }, Data: []byte(nil), User: "editorUser", @@ -828,6 +836,7 @@ func TestEventstore_Push(t *testing.T) { Type: "test.aggregate", ResourceOwner: "caos", InstanceID: "zitadel", + Version: "v1", }, Data: []byte(nil), User: "editorUser", @@ -883,6 +892,7 @@ func TestEventstore_Push(t *testing.T) { Type: "test.aggregate", ResourceOwner: "caos", InstanceID: "zitadel", + Version: "v1", }, Data: []byte(nil), User: "editorUser", diff --git a/internal/eventstore/local_crdb_test.go b/internal/eventstore/local_crdb_test.go index 6df9e9fd2..87c5084fe 100644 --- a/internal/eventstore/local_crdb_test.go +++ b/internal/eventstore/local_crdb_test.go @@ -9,6 +9,8 @@ import ( "time" "github.com/cockroachdb/cockroach-go/v2/testserver" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/jackc/pgx/v5/stdlib" "github.com/zitadel/logging" "github.com/zitadel/zitadel/cmd/initialise" @@ -39,10 +41,17 @@ func TestMain(m *testing.M) { testCRDBClient = &database.DB{ Database: new(testDB), } - testCRDBClient.DB, err = sql.Open("postgres", ts.PGURL().String()) + + connConfig, err := pgxpool.ParseConfig(ts.PGURL().String()) if err != nil { - logging.WithFields("error", err).Fatal("unable to connect to db") + logging.WithFields("error", err).Fatal("unable to parse db url") } + connConfig.AfterConnect = new_es.RegisterEventstoreTypes + pool, err := pgxpool.NewWithConfig(context.Background(), connConfig) + if err != nil { + logging.WithFields("error", err).Fatal("unable to create db pool") + } + testCRDBClient.DB = stdlib.OpenDBFromPool(pool) if err = testCRDBClient.Ping(); err != nil { logging.WithFields("error", err).Fatal("unable to ping db") } @@ -55,7 +64,7 @@ func TestMain(m *testing.M) { clients["v3(inmemory)"] = testCRDBClient if localDB, err := connectLocalhost(); err == nil { - if err = initDB(localDB); err != nil { + if err = initDB(context.Background(), localDB); err != nil { logging.WithFields("error", err).Fatal("migrations failed") } pushers["v3(singlenode)"] = new_es.NewEventstore(localDB) @@ -69,14 +78,14 @@ func TestMain(m *testing.M) { ts.Stop() }() - if err = initDB(testCRDBClient); err != nil { + if err = initDB(context.Background(), testCRDBClient); err != nil { logging.WithFields("error", err).Fatal("migrations failed") } os.Exit(m.Run()) } -func initDB(db *database.DB) error { +func initDB(ctx context.Context, db *database.DB) error { initialise.ReadStmts("cockroach") config := new(database.Config) config.SetConnector(&cockroach.Config{ @@ -85,7 +94,7 @@ func initDB(db *database.DB) error { }, Database: "zitadel", }) - err := initialise.Init(db, + err := initialise.Init(ctx, db, initialise.VerifyUser(config.Username(), ""), initialise.VerifyDatabase(config.DatabaseName()), initialise.VerifyGrant(config.DatabaseName(), config.Username()), @@ -93,7 +102,7 @@ func initDB(db *database.DB) error { if err != nil { return err } - err = initialise.VerifyZitadel(context.Background(), db, *config) + err = initialise.VerifyZitadel(ctx, db, *config) if err != nil { return err } diff --git a/internal/eventstore/repository/event.go b/internal/eventstore/repository/event.go index 57b85f15b..d0d2660d7 100644 --- a/internal/eventstore/repository/event.go +++ b/internal/eventstore/repository/event.go @@ -3,8 +3,12 @@ package repository import ( "database/sql" "encoding/json" + "strconv" + "strings" "time" + "github.com/zitadel/logging" + "github.com/zitadel/zitadel/internal/eventstore" ) @@ -82,7 +86,9 @@ func (e *Event) Type() eventstore.EventType { // Revision implements [eventstore.Event] func (e *Event) Revision() uint16 { - return 0 + revision, err := strconv.ParseUint(strings.TrimPrefix(string(e.Version), "v"), 10, 16) + logging.OnError(err).Debug("failed to parse event revision") + return uint16(revision) } // Sequence implements [eventstore.Event] diff --git a/internal/eventstore/repository/mock/repository.mock.go b/internal/eventstore/repository/mock/repository.mock.go index de04fef8c..8d5c0430a 100644 --- a/internal/eventstore/repository/mock/repository.mock.go +++ b/internal/eventstore/repository/mock/repository.mock.go @@ -165,7 +165,7 @@ func (mr *MockPusherMockRecorder) Health(arg0 any) *gomock.Call { } // Push mocks base method. -func (m *MockPusher) Push(arg0 context.Context, arg1 database.QueryExecuter, arg2 ...eventstore.Command) ([]eventstore.Event, error) { +func (m *MockPusher) Push(arg0 context.Context, arg1 database.ContextQueryExecuter, arg2 ...eventstore.Command) ([]eventstore.Event, error) { m.ctrl.T.Helper() varargs := []any{arg0, arg1} for _, a := range arg2 { diff --git a/internal/eventstore/repository/mock/repository.mock.impl.go b/internal/eventstore/repository/mock/repository.mock.impl.go index 365da7afe..9ae0b6b1e 100644 --- a/internal/eventstore/repository/mock/repository.mock.impl.go +++ b/internal/eventstore/repository/mock/repository.mock.impl.go @@ -80,7 +80,7 @@ func (m *MockRepository) ExpectInstanceIDsError(err error) *MockRepository { // The call will sleep at least the amount of passed duration. func (m *MockRepository) ExpectPush(expectedCommands []eventstore.Command, sleep time.Duration) *MockRepository { m.MockPusher.EXPECT().Push(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( - func(ctx context.Context, _ database.QueryExecuter, commands ...eventstore.Command) ([]eventstore.Event, error) { + func(ctx context.Context, _ database.ContextQueryExecuter, commands ...eventstore.Command) ([]eventstore.Event, error) { m.MockPusher.ctrl.T.Helper() time.Sleep(sleep) @@ -135,7 +135,7 @@ func (m *MockRepository) ExpectPushFailed(err error, expectedCommands []eventsto m.MockPusher.ctrl.T.Helper() m.MockPusher.EXPECT().Push(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( - func(ctx context.Context, _ database.QueryExecuter, commands ...eventstore.Command) ([]eventstore.Event, error) { + func(ctx context.Context, _ database.ContextQueryExecuter, commands ...eventstore.Command) ([]eventstore.Event, error) { if len(expectedCommands) != len(commands) { return nil, fmt.Errorf("unexpected amount of commands: want %d, got %d", len(expectedCommands), len(commands)) } @@ -197,7 +197,7 @@ func (e *mockEvent) CreatedAt() time.Time { func (m *MockRepository) ExpectRandomPush(expectedCommands []eventstore.Command) *MockRepository { m.MockPusher.EXPECT().Push(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( - func(ctx context.Context, _ database.QueryExecuter, commands ...eventstore.Command) ([]eventstore.Event, error) { + func(ctx context.Context, _ database.ContextQueryExecuter, commands ...eventstore.Command) ([]eventstore.Event, error) { assert.Len(m.MockPusher.ctrl.T, commands, len(expectedCommands)) events := make([]eventstore.Event, len(commands)) @@ -215,7 +215,7 @@ func (m *MockRepository) ExpectRandomPush(expectedCommands []eventstore.Command) func (m *MockRepository) ExpectRandomPushFailed(err error, expectedEvents []eventstore.Command) *MockRepository { m.MockPusher.EXPECT().Push(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( - func(ctx context.Context, _ database.QueryExecuter, events ...eventstore.Command) ([]eventstore.Event, error) { + func(ctx context.Context, _ database.ContextQueryExecuter, events ...eventstore.Command) ([]eventstore.Event, error) { assert.Len(m.MockPusher.ctrl.T, events, len(expectedEvents)) return nil, err }, diff --git a/internal/eventstore/repository/sql/local_crdb_test.go b/internal/eventstore/repository/sql/local_crdb_test.go index fccb16934..0f8c934b4 100644 --- a/internal/eventstore/repository/sql/local_crdb_test.go +++ b/internal/eventstore/repository/sql/local_crdb_test.go @@ -8,11 +8,14 @@ import ( "time" "github.com/cockroachdb/cockroach-go/v2/testserver" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/jackc/pgx/v5/stdlib" "github.com/zitadel/logging" "github.com/zitadel/zitadel/cmd/initialise" "github.com/zitadel/zitadel/internal/database" "github.com/zitadel/zitadel/internal/database/cockroach" + new_es "github.com/zitadel/zitadel/internal/eventstore/v3" ) var ( @@ -29,10 +32,18 @@ func TestMain(m *testing.M) { logging.WithFields("error", err).Fatal("unable to start db") } - testCRDBClient, err = sql.Open("postgres", ts.PGURL().String()) + connConfig, err := pgxpool.ParseConfig(ts.PGURL().String()) if err != nil { - logging.WithFields("error", err).Fatal("unable to connect to db") + logging.WithFields("error", err).Fatal("unable to parse db url") } + connConfig.AfterConnect = new_es.RegisterEventstoreTypes + pool, err := pgxpool.NewWithConfig(context.Background(), connConfig) + if err != nil { + logging.WithFields("error", err).Fatal("unable to create db pool") + } + + testCRDBClient = stdlib.OpenDBFromPool(pool) + if err = testCRDBClient.Ping(); err != nil { logging.WithFields("error", err).Fatal("unable to ping db") } @@ -42,14 +53,14 @@ func TestMain(m *testing.M) { ts.Stop() }() - if err = initDB(&database.DB{DB: testCRDBClient, Database: &cockroach.Config{Database: "zitadel"}}); err != nil { + if err = initDB(context.Background(), &database.DB{DB: testCRDBClient, Database: &cockroach.Config{Database: "zitadel"}}); err != nil { logging.WithFields("error", err).Fatal("migrations failed") } os.Exit(m.Run()) } -func initDB(db *database.DB) error { +func initDB(ctx context.Context, db *database.DB) error { config := new(database.Config) config.SetConnector(&cockroach.Config{User: cockroach.User{Username: "zitadel"}, Database: "zitadel"}) @@ -57,7 +68,7 @@ func initDB(db *database.DB) error { return err } - err := initialise.Init(db, + err := initialise.Init(ctx, db, initialise.VerifyUser(config.Username(), ""), initialise.VerifyDatabase(config.DatabaseName()), initialise.VerifyGrant(config.DatabaseName(), config.Username()), diff --git a/internal/eventstore/v3/event.go b/internal/eventstore/v3/event.go index e1c95f13f..da4e7a038 100644 --- a/internal/eventstore/v3/event.go +++ b/internal/eventstore/v3/event.go @@ -1,11 +1,13 @@ package eventstore import ( + "context" "encoding/json" + "strconv" "time" "github.com/zitadel/logging" - + "github.com/zitadel/zitadel/internal/api/authz" "github.com/zitadel/zitadel/internal/eventstore" "github.com/zitadel/zitadel/internal/zerrors" ) @@ -14,33 +16,98 @@ var ( _ eventstore.Event = (*event)(nil) ) +type command struct { + InstanceID string + AggregateType string + AggregateID string + CommandType string + Revision uint16 + Payload Payload + Creator string + Owner string +} + +func (c *command) Aggregate() *eventstore.Aggregate { + return &eventstore.Aggregate{ + ID: c.AggregateID, + Type: eventstore.AggregateType(c.AggregateType), + ResourceOwner: c.Owner, + InstanceID: c.InstanceID, + Version: eventstore.Version("v" + strconv.Itoa(int(c.Revision))), + } +} + type event struct { - aggregate *eventstore.Aggregate - creator string - revision uint16 - typ eventstore.EventType + command *command createdAt time.Time sequence uint64 position float64 - payload Payload } -func commandToEvent(sequence *latestSequence, command eventstore.Command) (_ *event, err error) { +// TODO: remove on v3 +func commandToEventOld(sequence *latestSequence, cmd eventstore.Command) (_ *event, err error) { var payload Payload - if command.Payload() != nil { - payload, err = json.Marshal(command.Payload()) + if cmd.Payload() != nil { + payload, err = json.Marshal(cmd.Payload()) if err != nil { logging.WithError(err).Warn("marshal payload failed") return nil, zerrors.ThrowInternal(err, "V3-MInPK", "Errors.Internal") } } return &event{ - aggregate: sequence.aggregate, - creator: command.Creator(), - revision: command.Revision(), - typ: command.Type(), - payload: payload, - sequence: sequence.sequence, + command: &command{ + InstanceID: sequence.aggregate.InstanceID, + AggregateType: string(sequence.aggregate.Type), + AggregateID: sequence.aggregate.ID, + CommandType: string(cmd.Type()), + Revision: cmd.Revision(), + Payload: payload, + Creator: cmd.Creator(), + Owner: sequence.aggregate.ResourceOwner, + }, + sequence: sequence.sequence, + }, nil +} + +func commandsToEvents(ctx context.Context, cmds []eventstore.Command) (_ []eventstore.Event, _ []*command, err error) { + events := make([]eventstore.Event, len(cmds)) + commands := make([]*command, len(cmds)) + for i, cmd := range cmds { + if cmd.Aggregate().InstanceID == "" { + cmd.Aggregate().InstanceID = authz.GetInstance(ctx).InstanceID() + } + events[i], err = commandToEvent(cmd) + if err != nil { + return nil, nil, err + } + commands[i] = events[i].(*event).command + } + return events, commands, nil +} + +func commandToEvent(cmd eventstore.Command) (_ eventstore.Event, err error) { + var payload Payload + if cmd.Payload() != nil { + payload, err = json.Marshal(cmd.Payload()) + if err != nil { + logging.WithError(err).Warn("marshal payload failed") + return nil, zerrors.ThrowInternal(err, "V3-MInPK", "Errors.Internal") + } + } + + command := &command{ + InstanceID: cmd.Aggregate().InstanceID, + AggregateType: string(cmd.Aggregate().Type), + AggregateID: cmd.Aggregate().ID, + CommandType: string(cmd.Type()), + Revision: cmd.Revision(), + Payload: payload, + Creator: cmd.Creator(), + Owner: cmd.Aggregate().ResourceOwner, + } + + return &event{ + command: command, }, nil } @@ -56,22 +123,22 @@ func (e *event) EditorUser() string { // Aggregate implements [eventstore.Event] func (e *event) Aggregate() *eventstore.Aggregate { - return e.aggregate + return e.command.Aggregate() } // Creator implements [eventstore.Event] func (e *event) Creator() string { - return e.creator + return e.command.Creator } // Revision implements [eventstore.Event] func (e *event) Revision() uint16 { - return e.revision + return e.command.Revision } // Type implements [eventstore.Event] func (e *event) Type() eventstore.EventType { - return e.typ + return eventstore.EventType(e.command.CommandType) } // CreatedAt implements [eventstore.Event] @@ -91,10 +158,10 @@ func (e *event) Position() float64 { // Unmarshal implements [eventstore.Event] func (e *event) Unmarshal(ptr any) error { - if len(e.payload) == 0 { + if len(e.command.Payload) == 0 { return nil } - if err := json.Unmarshal(e.payload, ptr); err != nil { + if err := json.Unmarshal(e.command.Payload, ptr); err != nil { return zerrors.ThrowInternal(err, "V3-u8qVo", "Errors.Internal") } @@ -103,5 +170,5 @@ func (e *event) Unmarshal(ptr any) error { // DataAsBytes implements [eventstore.Event] func (e *event) DataAsBytes() []byte { - return e.payload + return e.command.Payload } diff --git a/internal/eventstore/v3/event_test.go b/internal/eventstore/v3/event_test.go index 82a3aa1c9..bd813c6ed 100644 --- a/internal/eventstore/v3/event_test.go +++ b/internal/eventstore/v3/event_test.go @@ -1,16 +1,122 @@ package eventstore import ( + "context" "encoding/json" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/zitadel/zitadel/internal/api/authz" "github.com/zitadel/zitadel/internal/eventstore" ) func Test_commandToEvent(t *testing.T) { + payload := struct { + ID string + }{ + ID: "test", + } + payloadMarshalled, err := json.Marshal(payload) + if err != nil { + t.Fatalf("marshal of payload failed: %v", err) + } + type args struct { + command eventstore.Command + } + type want struct { + event *event + err func(t *testing.T, err error) + } + tests := []struct { + name string + args args + want want + }{ + { + name: "no payload", + args: args{ + command: &mockCommand{ + aggregate: mockAggregate("V3-Red9I"), + payload: nil, + }, + }, + want: want{ + event: mockEvent( + mockAggregate("V3-Red9I"), + 0, + nil, + ).(*event), + }, + }, + { + name: "struct payload", + args: args{ + command: &mockCommand{ + aggregate: mockAggregate("V3-Red9I"), + payload: payload, + }, + }, + want: want{ + event: mockEvent( + mockAggregate("V3-Red9I"), + 0, + payloadMarshalled, + ).(*event), + }, + }, + { + name: "pointer payload", + args: args{ + command: &mockCommand{ + aggregate: mockAggregate("V3-Red9I"), + payload: &payload, + }, + }, + want: want{ + event: mockEvent( + mockAggregate("V3-Red9I"), + 0, + payloadMarshalled, + ).(*event), + }, + }, + { + name: "invalid payload", + args: args{ + command: &mockCommand{ + aggregate: mockAggregate("V3-Red9I"), + payload: func() {}, + }, + }, + want: want{ + err: func(t *testing.T, err error) { + assert.Error(t, err) + }, + }, + }, + } + for _, tt := range tests { + if tt.want.err == nil { + tt.want.err = func(t *testing.T, err error) { + require.NoError(t, err) + } + } + t.Run(tt.name, func(t *testing.T) { + got, err := commandToEvent(tt.args.command) + + tt.want.err(t, err) + if tt.want.event == nil { + assert.Nil(t, got) + return + } + assert.Equal(t, tt.want.event, got) + }) + } +} + +func Test_commandToEventOld(t *testing.T) { payload := struct { ID string }{ @@ -119,10 +225,258 @@ func Test_commandToEvent(t *testing.T) { } } t.Run(tt.name, func(t *testing.T) { - got, err := commandToEvent(tt.args.sequence, tt.args.command) + got, err := commandToEventOld(tt.args.sequence, tt.args.command) tt.want.err(t, err) assert.Equal(t, tt.want.event, got) }) } } + +func Test_commandsToEvents(t *testing.T) { + ctx := context.Background() + payload := struct { + ID string + }{ + ID: "test", + } + payloadMarshalled, err := json.Marshal(payload) + if err != nil { + t.Fatalf("marshal of payload failed: %v", err) + } + type args struct { + ctx context.Context + cmds []eventstore.Command + } + type want struct { + events []eventstore.Event + commands []*command + err func(t *testing.T, err error) + } + tests := []struct { + name string + args args + want want + }{ + { + name: "no commands", + args: args{ + ctx: ctx, + cmds: nil, + }, + want: want{ + events: []eventstore.Event{}, + commands: []*command{}, + err: func(t *testing.T, err error) { + require.NoError(t, err) + }, + }, + }, + { + name: "single command no payload", + args: args{ + ctx: ctx, + cmds: []eventstore.Command{ + &mockCommand{ + aggregate: mockAggregate("V3-Red9I"), + payload: nil, + }, + }, + }, + want: want{ + events: []eventstore.Event{ + mockEvent( + mockAggregate("V3-Red9I"), + 0, + nil, + ), + }, + commands: []*command{ + { + InstanceID: "instance", + AggregateType: "type", + AggregateID: "V3-Red9I", + Owner: "ro", + CommandType: "event.type", + Revision: 1, + Payload: nil, + Creator: "creator", + }, + }, + err: func(t *testing.T, err error) { + require.NoError(t, err) + }, + }, + }, + { + name: "single command no instance id", + args: args{ + ctx: authz.WithInstanceID(ctx, "instance from ctx"), + cmds: []eventstore.Command{ + &mockCommand{ + aggregate: mockAggregateWithInstance("V3-Red9I", ""), + payload: nil, + }, + }, + }, + want: want{ + events: []eventstore.Event{ + mockEvent( + mockAggregateWithInstance("V3-Red9I", "instance from ctx"), + 0, + nil, + ), + }, + commands: []*command{ + { + InstanceID: "instance from ctx", + AggregateType: "type", + AggregateID: "V3-Red9I", + Owner: "ro", + CommandType: "event.type", + Revision: 1, + Payload: nil, + Creator: "creator", + }, + }, + err: func(t *testing.T, err error) { + require.NoError(t, err) + }, + }, + }, + { + name: "single command with payload", + args: args{ + ctx: ctx, + cmds: []eventstore.Command{ + &mockCommand{ + aggregate: mockAggregate("V3-Red9I"), + payload: payload, + }, + }, + }, + want: want{ + events: []eventstore.Event{ + mockEvent( + mockAggregate("V3-Red9I"), + 0, + payloadMarshalled, + ), + }, + commands: []*command{ + { + InstanceID: "instance", + AggregateType: "type", + AggregateID: "V3-Red9I", + Owner: "ro", + CommandType: "event.type", + Revision: 1, + Payload: payloadMarshalled, + Creator: "creator", + }, + }, + err: func(t *testing.T, err error) { + require.NoError(t, err) + }, + }, + }, + { + name: "multiple commands", + args: args{ + ctx: ctx, + cmds: []eventstore.Command{ + &mockCommand{ + aggregate: mockAggregate("V3-Red9I"), + payload: payload, + }, + &mockCommand{ + aggregate: mockAggregate("V3-Red9I"), + payload: nil, + }, + }, + }, + want: want{ + events: []eventstore.Event{ + mockEvent( + mockAggregate("V3-Red9I"), + 0, + payloadMarshalled, + ), + mockEvent( + mockAggregate("V3-Red9I"), + 0, + nil, + ), + }, + commands: []*command{ + { + InstanceID: "instance", + AggregateType: "type", + AggregateID: "V3-Red9I", + CommandType: "event.type", + Revision: 1, + Payload: payloadMarshalled, + Creator: "creator", + Owner: "ro", + }, + { + InstanceID: "instance", + AggregateType: "type", + AggregateID: "V3-Red9I", + CommandType: "event.type", + Revision: 1, + Payload: nil, + Creator: "creator", + Owner: "ro", + }, + }, + err: func(t *testing.T, err error) { + require.NoError(t, err) + }, + }, + }, + { + name: "invalid command", + args: args{ + ctx: ctx, + cmds: []eventstore.Command{ + &mockCommand{ + aggregate: mockAggregate("V3-Red9I"), + payload: func() {}, + }, + }, + }, + want: want{ + events: nil, + commands: nil, + err: func(t *testing.T, err error) { + assert.Error(t, err) + }, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotEvents, gotCommands, err := commandsToEvents(tt.args.ctx, tt.args.cmds) + + tt.want.err(t, err) + assert.Equal(t, tt.want.events, gotEvents) + require.Len(t, gotCommands, len(tt.want.commands)) + for i, wantCommand := range tt.want.commands { + assertCommand(t, wantCommand, gotCommands[i]) + } + }) + } +} + +func assertCommand(t *testing.T, want, got *command) { + t.Helper() + assert.Equal(t, want.CommandType, got.CommandType) + assert.Equal(t, want.Payload, got.Payload) + assert.Equal(t, want.Creator, got.Creator) + assert.Equal(t, want.Owner, got.Owner) + assert.Equal(t, want.AggregateID, got.AggregateID) + assert.Equal(t, want.AggregateType, got.AggregateType) + assert.Equal(t, want.InstanceID, got.InstanceID) + assert.Equal(t, want.Revision, got.Revision) +} diff --git a/internal/eventstore/v3/eventstore.go b/internal/eventstore/v3/eventstore.go index 7c58f53f2..1bb515527 100644 --- a/internal/eventstore/v3/eventstore.go +++ b/internal/eventstore/v3/eventstore.go @@ -2,11 +2,26 @@ package eventstore import ( "context" + "database/sql" + "encoding/json" + "errors" + "sync" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/stdlib" + "github.com/zitadel/logging" "github.com/zitadel/zitadel/internal/database" + "github.com/zitadel/zitadel/internal/database/dialect" "github.com/zitadel/zitadel/internal/eventstore" ) +func init() { + dialect.RegisterAfterConnect(RegisterEventstoreTypes) +} + var ( // pushPlaceholderFmt defines how data are inserted into the events table pushPlaceholderFmt string @@ -20,6 +35,123 @@ type Eventstore struct { client *database.DB } +var ( + textType = &pgtype.Type{ + Name: "text", + OID: pgtype.TextOID, + Codec: pgtype.TextCodec{}, + } + commandType = &pgtype.Type{ + Codec: &pgtype.CompositeCodec{ + Fields: []pgtype.CompositeCodecField{ + { + Name: "instance_id", + Type: textType, + }, + { + Name: "aggregate_type", + Type: textType, + }, + { + Name: "aggregate_id", + Type: textType, + }, + { + Name: "command_type", + Type: textType, + }, + { + Name: "revision", + Type: &pgtype.Type{ + Name: "int2", + OID: pgtype.Int2OID, + Codec: pgtype.Int2Codec{}, + }, + }, + { + Name: "payload", + Type: &pgtype.Type{ + Name: "jsonb", + OID: pgtype.JSONBOID, + Codec: &pgtype.JSONBCodec{ + Marshal: json.Marshal, + Unmarshal: json.Unmarshal, + }, + }, + }, + { + Name: "creator", + Type: textType, + }, + { + Name: "owner", + Type: textType, + }, + }, + }, + } + commandArrayCodec = &pgtype.Type{ + Codec: &pgtype.ArrayCodec{ + ElementType: commandType, + }, + } +) + +var typeMu sync.Mutex + +func RegisterEventstoreTypes(ctx context.Context, conn *pgx.Conn) error { + // conn.TypeMap is not thread safe + typeMu.Lock() + defer typeMu.Unlock() + + m := conn.TypeMap() + + var cmd *command + if _, ok := m.TypeForValue(cmd); ok { + return nil + } + + if commandType.OID == 0 || commandArrayCodec.OID == 0 { + err := conn.QueryRow(ctx, "select oid, typarray from pg_type where typname = $1 and typnamespace = (select oid from pg_namespace where nspname = $2)", "command", "eventstore"). + Scan(&commandType.OID, &commandArrayCodec.OID) + if err != nil { + logging.WithError(err).Debug("failed to get oid for command type") + return nil + } + if commandType.OID == 0 || commandArrayCodec.OID == 0 { + logging.Debug("oid for command type not found") + return nil + } + } + + m.RegisterTypes([]*pgtype.Type{ + { + Name: "eventstore.command", + Codec: commandType.Codec, + OID: commandType.OID, + }, + { + Name: "command", + Codec: commandType.Codec, + OID: commandType.OID, + }, + { + Name: "eventstore._command", + Codec: commandArrayCodec.Codec, + OID: commandArrayCodec.OID, + }, + { + Name: "_command", + Codec: commandArrayCodec.Codec, + OID: commandArrayCodec.OID, + }, + }) + dialect.RegisterDefaultPgTypeVariants[command](m, "eventstore.command", "eventstore._command") + dialect.RegisterDefaultPgTypeVariants[command](m, "command", "_command") + + return nil +} + // Client implements the [eventstore.Pusher] func (es *Eventstore) Client() *database.DB { return es.client @@ -41,3 +173,45 @@ func NewEventstore(client *database.DB) *Eventstore { func (es *Eventstore) Health(ctx context.Context) error { return es.client.PingContext(ctx) } + +var errTypesNotFound = errors.New("types not found") + +func CheckExecutionPlan(ctx context.Context, conn *sql.Conn) error { + return conn.Raw(func(driverConn any) error { + if _, ok := driverConn.(sqlmock.SqlmockCommon); ok { + return nil + } + conn, ok := driverConn.(*stdlib.Conn) + if !ok { + return errTypesNotFound + } + + return RegisterEventstoreTypes(ctx, conn.Conn()) + }) +} + +func (es *Eventstore) pushTx(ctx context.Context, client database.ContextQueryExecuter) (tx database.Tx, deferrable func(err error) error, err error) { + tx, ok := client.(database.Tx) + if ok { + return tx, nil, nil + } + beginner, ok := client.(database.Beginner) + if !ok { + beginner = es.client + } + + isolationLevel := sql.LevelReadCommitted + // cockroach requires serializable to execute the push function + // because we use [cluster_logical_timestamp()](https://www.cockroachlabs.com/docs/stable/functions-and-operators#system-info-functions) + if es.client.Type() == "cockroach" { + isolationLevel = sql.LevelSerializable + } + tx, err = beginner.BeginTx(ctx, &sql.TxOptions{ + Isolation: isolationLevel, + ReadOnly: false, + }) + if err != nil { + return nil, nil, err + } + return tx, func(err error) error { return database.CloseTransaction(tx, err) }, nil +} diff --git a/internal/eventstore/v3/field.go b/internal/eventstore/v3/field.go index cfa9c08bb..b399e7f5e 100644 --- a/internal/eventstore/v3/field.go +++ b/internal/eventstore/v3/field.go @@ -143,7 +143,7 @@ func buildSearchCondition(builder *strings.Builder, index int, conditions map[ev return args } -func handleFieldCommands(ctx context.Context, tx database.Tx, commands []eventstore.Command) error { +func (es *Eventstore) handleFieldCommands(ctx context.Context, tx database.Tx, commands []eventstore.Command) error { for _, command := range commands { if len(command.Fields()) > 0 { if err := handleFieldOperations(ctx, tx, command.Fields()); err != nil { diff --git a/internal/eventstore/v3/mock_test.go b/internal/eventstore/v3/mock_test.go index 8de5ae8e2..89a50f8fd 100644 --- a/internal/eventstore/v3/mock_test.go +++ b/internal/eventstore/v3/mock_test.go @@ -48,12 +48,17 @@ func (e *mockCommand) Fields() []*eventstore.FieldOperation { func mockEvent(aggregate *eventstore.Aggregate, sequence uint64, payload Payload) eventstore.Event { return &event{ - aggregate: aggregate, - creator: "creator", - revision: 1, - typ: "event.type", - sequence: sequence, - payload: payload, + command: &command{ + InstanceID: aggregate.InstanceID, + AggregateType: string(aggregate.Type), + AggregateID: aggregate.ID, + Owner: aggregate.ResourceOwner, + Creator: "creator", + Revision: 1, + CommandType: "event.type", + Payload: payload, + }, + sequence: sequence, } } @@ -66,3 +71,13 @@ func mockAggregate(id string) *eventstore.Aggregate { Version: "v1", } } + +func mockAggregateWithInstance(id, instance string) *eventstore.Aggregate { + return &eventstore.Aggregate{ + ID: id, + InstanceID: instance, + Type: "type", + ResourceOwner: "ro", + Version: "v1", + } +} diff --git a/internal/eventstore/v3/push.go b/internal/eventstore/v3/push.go index c0f66209c..fb597021e 100644 --- a/internal/eventstore/v3/push.go +++ b/internal/eventstore/v3/push.go @@ -4,83 +4,58 @@ import ( "context" "database/sql" _ "embed" - "errors" - "fmt" - "strconv" - "strings" - "github.com/cockroachdb/cockroach-go/v2/crdb" - "github.com/jackc/pgx/v5/pgconn" "github.com/zitadel/logging" - "github.com/zitadel/zitadel/internal/api/authz" "github.com/zitadel/zitadel/internal/database" - "github.com/zitadel/zitadel/internal/database/dialect" "github.com/zitadel/zitadel/internal/eventstore" "github.com/zitadel/zitadel/internal/telemetry/tracing" - "github.com/zitadel/zitadel/internal/zerrors" ) -var appNamePrefix = dialect.DBPurposeEventPusher.AppName() + "_" - var pushTxOpts = &sql.TxOptions{ Isolation: sql.LevelReadCommitted, ReadOnly: false, } -func (es *Eventstore) Push(ctx context.Context, client database.QueryExecuter, commands ...eventstore.Command) (events []eventstore.Event, err error) { +func (es *Eventstore) Push(ctx context.Context, client database.ContextQueryExecuter, commands ...eventstore.Command) (events []eventstore.Event, err error) { ctx, span := tracing.NewSpan(ctx) defer func() { span.EndWithError(err) }() - var tx database.Tx + events, err = es.writeCommands(ctx, client, commands) + if isSetupNotExecutedError(err) { + return es.pushWithoutFunc(ctx, client, commands...) + } + + return events, err +} + +func (es *Eventstore) writeCommands(ctx context.Context, client database.ContextQueryExecuter, commands []eventstore.Command) (_ []eventstore.Event, err error) { + var conn *sql.Conn switch c := client.(type) { - case database.Tx: - tx = c case database.Client: - // We cannot use READ COMMITTED on CockroachDB because we use cluster_logical_timestamp() which is not supported in this isolation level - var opts *sql.TxOptions - if es.client.Database.Type() == "postgres" { - opts = pushTxOpts - } - tx, err = c.BeginTx(ctx, opts) - if err != nil { - return nil, err - } - defer func() { - err = database.CloseTransaction(tx, err) - }() - default: - // We cannot use READ COMMITTED on CockroachDB because we use cluster_logical_timestamp() which is not supported in this isolation level - var opts *sql.TxOptions - if es.client.Database.Type() == "postgres" { - opts = pushTxOpts - } - tx, err = es.client.BeginTx(ctx, opts) - if err != nil { - return nil, err - } - defer func() { - err = database.CloseTransaction(tx, err) - }() + conn, err = c.Conn(ctx) + case nil: + conn, err = es.client.Conn(ctx) + client = conn } - // tx is not closed because [crdb.ExecuteInTx] takes care of that - var ( - sequences []*latestSequence - ) - - // needs to be set like this because psql complains about parameters in the SET statement - _, err = tx.ExecContext(ctx, "SET application_name = '"+appNamePrefix+authz.GetInstance(ctx).InstanceID()+"'") - if err != nil { - logging.WithError(err).Warn("failed to set application name") - return nil, err - } - - sequences, err = latestSequences(ctx, tx, commands) if err != nil { return nil, err } + if conn != nil { + defer conn.Close() + } - events, err = insertEvents(ctx, tx, sequences, commands) + tx, close, err := es.pushTx(ctx, client) + if err != nil { + return nil, err + } + if close != nil { + defer func() { + err = close(err) + }() + } + + events, err := writeEvents(ctx, tx, commands) if err != nil { return nil, err } @@ -89,16 +64,7 @@ func (es *Eventstore) Push(ctx context.Context, client database.QueryExecuter, c return nil, 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 nil, err - } - } - - err = handleFieldCommands(ctx, tx, commands) + err = es.handleFieldCommands(ctx, tx, commands) if err != nil { return nil, err } @@ -106,120 +72,30 @@ func (es *Eventstore) Push(ctx context.Context, client database.QueryExecuter, c return events, nil } -//go:embed push.sql -var pushStmt string +func writeEvents(ctx context.Context, tx database.Tx, commands []eventstore.Command) (_ []eventstore.Event, err error) { + ctx, span := tracing.NewSpan(ctx) + defer func() { span.EndWithError(err) }() -func insertEvents(ctx context.Context, tx database.Tx, sequences []*latestSequence, commands []eventstore.Command) ([]eventstore.Event, error) { - events, placeholders, args, err := mapCommands(commands, sequences) + events, cmds, err := commandsToEvents(ctx, commands) if err != nil { return nil, err } - rows, err := tx.QueryContext(ctx, fmt.Sprintf(pushStmt, strings.Join(placeholders, ", ")), args...) + rows, err := tx.QueryContext(ctx, `select owner, created_at, "sequence", position from eventstore.push($1::eventstore.command[])`, cmds) if err != nil { return nil, err } defer rows.Close() for i := 0; rows.Next(); i++ { - err = rows.Scan(&events[i].(*event).createdAt, &events[i].(*event).position) + err = rows.Scan(&events[i].(*event).command.Owner, &events[i].(*event).createdAt, &events[i].(*event).sequence, &events[i].(*event).position) if err != nil { logging.WithError(err).Warn("failed to scan events") return nil, err } } - - if err := rows.Err(); err != nil { - pgErr := new(pgconn.PgError) - if errors.As(err, &pgErr) { - // Check if push tries to write an event just written - // by another transaction - if pgErr.Code == "40001" { - // TODO: @livio-a should we return the parent or not? - return nil, zerrors.ThrowInvalidArgument(err, "V3-p5xAn", "Errors.AlreadyExists") - } - } - logging.WithError(rows.Err()).Warn("failed to push events") - return nil, zerrors.ThrowInternal(err, "V3-VGnZY", "Errors.Internal") + if err = rows.Err(); err != nil { + return nil, err } - return events, nil } - -const argsPerCommand = 10 - -func mapCommands(commands []eventstore.Command, sequences []*latestSequence) (events []eventstore.Event, placeholders []string, args []any, err error) { - events = make([]eventstore.Event, len(commands)) - args = make([]any, 0, len(commands)*argsPerCommand) - placeholders = make([]string, len(commands)) - - for i, command := range commands { - sequence := searchSequenceByCommand(sequences, command) - if sequence == nil { - logging.WithFields( - "aggType", command.Aggregate().Type, - "aggID", command.Aggregate().ID, - "instance", command.Aggregate().InstanceID, - ).Panic("no sequence found") - // added return for linting - return nil, nil, nil, nil - } - sequence.sequence++ - - events[i], err = commandToEvent(sequence, command) - if err != nil { - return nil, nil, nil, err - } - - placeholders[i] = fmt.Sprintf(pushPlaceholderFmt, - i*argsPerCommand+1, - i*argsPerCommand+2, - i*argsPerCommand+3, - i*argsPerCommand+4, - i*argsPerCommand+5, - i*argsPerCommand+6, - i*argsPerCommand+7, - i*argsPerCommand+8, - i*argsPerCommand+9, - i*argsPerCommand+10, - ) - - revision, err := strconv.Atoi(strings.TrimPrefix(string(events[i].(*event).aggregate.Version), "v")) - if err != nil { - return nil, nil, nil, zerrors.ThrowInternal(err, "V3-JoZEp", "Errors.Internal") - } - args = append(args, - events[i].(*event).aggregate.InstanceID, - events[i].(*event).aggregate.ResourceOwner, - events[i].(*event).aggregate.Type, - events[i].(*event).aggregate.ID, - revision, - events[i].(*event).creator, - events[i].(*event).typ, - events[i].(*event).payload, - events[i].(*event).sequence, - i, - ) - } - - return events, placeholders, args, nil -} - -type transaction struct { - database.Tx -} - -var _ crdb.Tx = (*transaction)(nil) - -func (t *transaction) Exec(ctx context.Context, query string, args ...interface{}) error { - _, err := t.Tx.ExecContext(ctx, query, args...) - return err -} - -func (t *transaction) Commit(ctx context.Context) error { - return t.Tx.Commit() -} - -func (t *transaction) Rollback(ctx context.Context) error { - return t.Tx.Rollback() -} diff --git a/internal/eventstore/v3/push_test.go b/internal/eventstore/v3/push_test.go index 0f16a2ac7..a6c4f515f 100644 --- a/internal/eventstore/v3/push_test.go +++ b/internal/eventstore/v3/push_test.go @@ -70,11 +70,11 @@ func Test_mapCommands(t *testing.T) { args: []any{ "instance", "ro", - eventstore.AggregateType("type"), + "type", "V3-VEIvq", - 1, + uint16(1), "creator", - eventstore.EventType("event.type"), + "event.type", Payload(nil), uint64(1), 0, @@ -121,22 +121,22 @@ func Test_mapCommands(t *testing.T) { // first event "instance", "ro", - eventstore.AggregateType("type"), + "type", "V3-VEIvq", - 1, + uint16(1), "creator", - eventstore.EventType("event.type"), + "event.type", Payload(nil), uint64(6), 0, // second event "instance", "ro", - eventstore.AggregateType("type"), + "type", "V3-VEIvq", - 1, + uint16(1), "creator", - eventstore.EventType("event.type"), + "event.type", Payload(nil), uint64(7), 1, @@ -187,22 +187,22 @@ func Test_mapCommands(t *testing.T) { // first event "instance", "ro", - eventstore.AggregateType("type"), + "type", "V3-VEIvq", - 1, + uint16(1), "creator", - eventstore.EventType("event.type"), + "event.type", Payload(nil), uint64(6), 0, // second event "instance", "ro", - eventstore.AggregateType("type"), + "type", "V3-IT6VN", - 1, + uint16(1), "creator", - eventstore.EventType("event.type"), + "event.type", Payload(nil), uint64(1), 1, diff --git a/internal/eventstore/v3/push_without_func.go b/internal/eventstore/v3/push_without_func.go new file mode 100644 index 000000000..914b88020 --- /dev/null +++ b/internal/eventstore/v3/push_without_func.go @@ -0,0 +1,183 @@ +package eventstore + +import ( + "context" + _ "embed" + "errors" + "fmt" + "strings" + + "github.com/cockroachdb/cockroach-go/v2/crdb" + "github.com/jackc/pgx/v5/pgconn" + "github.com/zitadel/logging" + + "github.com/zitadel/zitadel/internal/database" + "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/zerrors" +) + +type transaction struct { + database.Tx +} + +var _ crdb.Tx = (*transaction)(nil) + +func (t *transaction) Exec(ctx context.Context, query string, args ...interface{}) error { + _, err := t.Tx.ExecContext(ctx, query, args...) + return err +} + +func (t *transaction) Commit(ctx context.Context) error { + return t.Tx.Commit() +} + +func (t *transaction) Rollback(ctx context.Context) error { + return t.Tx.Rollback() +} + +// checks whether the error is caused because setup step 39 was not executed +func isSetupNotExecutedError(err error) bool { + if err == nil { + return false + } + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) { + return (pgErr.Code == "42704" && strings.Contains(pgErr.Message, "eventstore.command")) || + (pgErr.Code == "42883" && strings.Contains(pgErr.Message, "eventstore.push")) + } + return errors.Is(err, errTypesNotFound) +} + +var ( + //go:embed push.sql + pushStmt string +) + +// pushWithoutFunc implements pushing events before setup step 39 was introduced. +// TODO: remove with v3 +func (es *Eventstore) pushWithoutFunc(ctx context.Context, client database.ContextQueryExecuter, commands ...eventstore.Command) (events []eventstore.Event, err error) { + tx, closeTx, err := es.pushTx(ctx, client) + if err != nil { + return nil, err + } + defer func() { + err = closeTx(err) + }() + + // tx is not closed because [crdb.ExecuteInTx] takes care of that + var ( + sequences []*latestSequence + ) + sequences, err = latestSequences(ctx, tx, commands) + if err != nil { + return nil, err + } + + events, err = es.writeEventsOld(ctx, tx, sequences, commands) + if err != nil { + return nil, err + } + + if err = handleUniqueConstraints(ctx, tx, commands); err != nil { + return nil, err + } + + err = es.handleFieldCommands(ctx, tx, commands) + if err != nil { + return nil, err + } + + return events, nil +} + +func (es *Eventstore) writeEventsOld(ctx context.Context, tx database.Tx, sequences []*latestSequence, commands []eventstore.Command) ([]eventstore.Event, error) { + events, placeholders, args, err := mapCommands(commands, sequences) + if err != nil { + return nil, err + } + + rows, err := tx.QueryContext(ctx, fmt.Sprintf(pushStmt, strings.Join(placeholders, ", ")), args...) + if err != nil { + return nil, err + } + defer rows.Close() + + for i := 0; rows.Next(); i++ { + err = rows.Scan(&events[i].(*event).createdAt, &events[i].(*event).position) + if err != nil { + logging.WithError(err).Warn("failed to scan events") + return nil, err + } + } + + if err := rows.Err(); err != nil { + pgErr := new(pgconn.PgError) + if errors.As(err, &pgErr) { + // Check if push tries to write an event just written + // by another transaction + if pgErr.Code == "40001" { + // TODO: @livio-a should we return the parent or not? + return nil, zerrors.ThrowInvalidArgument(err, "V3-p5xAn", "Errors.AlreadyExists") + } + } + logging.WithError(rows.Err()).Warn("failed to push events") + return nil, zerrors.ThrowInternal(err, "V3-VGnZY", "Errors.Internal") + } + + return events, nil +} + +const argsPerCommand = 10 + +func mapCommands(commands []eventstore.Command, sequences []*latestSequence) (events []eventstore.Event, placeholders []string, args []any, err error) { + events = make([]eventstore.Event, len(commands)) + args = make([]any, 0, len(commands)*argsPerCommand) + placeholders = make([]string, len(commands)) + + for i, command := range commands { + sequence := searchSequenceByCommand(sequences, command) + if sequence == nil { + logging.WithFields( + "aggType", command.Aggregate().Type, + "aggID", command.Aggregate().ID, + "instance", command.Aggregate().InstanceID, + ).Panic("no sequence found") + // added return for linting + return nil, nil, nil, nil + } + sequence.sequence++ + + events[i], err = commandToEventOld(sequence, command) + if err != nil { + return nil, nil, nil, err + } + + placeholders[i] = fmt.Sprintf(pushPlaceholderFmt, + i*argsPerCommand+1, + i*argsPerCommand+2, + i*argsPerCommand+3, + i*argsPerCommand+4, + i*argsPerCommand+5, + i*argsPerCommand+6, + i*argsPerCommand+7, + i*argsPerCommand+8, + i*argsPerCommand+9, + i*argsPerCommand+10, + ) + + args = append(args, + events[i].(*event).command.InstanceID, + events[i].(*event).command.Owner, + events[i].(*event).command.AggregateType, + events[i].(*event).command.AggregateID, + events[i].(*event).command.Revision, + events[i].(*event).command.Creator, + events[i].(*event).command.CommandType, + events[i].(*event).command.Payload, + events[i].(*event).sequence, + i, + ) + } + + return events, placeholders, args, nil +} diff --git a/internal/eventstore/v3/unique_constraints.go b/internal/eventstore/v3/unique_constraints.go index 9c4d1831c..a491ae4f5 100644 --- a/internal/eventstore/v3/unique_constraints.go +++ b/internal/eventstore/v3/unique_constraints.go @@ -12,6 +12,7 @@ import ( "github.com/zitadel/zitadel/internal/database" "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/telemetry/tracing" "github.com/zitadel/zitadel/internal/zerrors" ) @@ -24,7 +25,10 @@ var ( addConstraintStmt string ) -func handleUniqueConstraints(ctx context.Context, tx database.Tx, commands []eventstore.Command) error { +func handleUniqueConstraints(ctx context.Context, tx database.Tx, commands []eventstore.Command) (err error) { + ctx, span := tracing.NewSpan(ctx) + defer func() { span.EndWithError(err) }() + deletePlaceholders := make([]string, 0) deleteArgs := make([]any, 0)