Files
zitadel/cmd/setup/setup.go
T
684de2e691 perf(query): use UNION of index seeks for ListUsers login equality (#12701)
<!-- CURSOR_AGENT_PR_BODY_BEGIN -->
# Which Problems Are Solved

Login V2 `ListUsers` lookups of the form username OR email (and email OR
phone) hash-join the full `users14` / `users14_humans` tables, then
filter. That cost grows with instance size (~40ms at 10k users, ~650ms
at 200k).

Top-level login-name equals already used an indexed join, but the same
login-name filter inside an `OrQuery` fell back to the slow view-based
exists query.

# How the Problems Are Solved

**TL;DR:** equality lookups on username, email, phone, or login name
become `INNER JOIN (UNION of indexed ID seeks) AS matches`. Mixed /
contains / name searches keep the old WHERE shape.

### Reviewer map

| Read first | Why |
| --- | --- |
| `internal/query/user_login_equality.go` | Typed filters + extractor +
UNION join |
| `internal/query/user.go` (`prepareUsersQuery`, constructors) | Wiring;
username/email/phone equals return `lowerEqualsFilter` |
| `cmd/setup/76.go` + `cmd/setup/76/*.sql` | Concurrent indexes for
existing deployments; drop invalid leftovers |
| `internal/query/projection/user.go` | Same indexes in `Init()` for new
databases |
| `internal/query/user_login_name_exists_test.go` | SQL-shape coverage |
| `internal/api/grpc/user/v2/integration_test/query_test.go` | Nested
username AND/OR user results |

### Query rewrite

- Extract one equals leaf, a flat `OrQuery` of those leaves, or the same
nested under `AndQuery`. Remaining AND conjuncts stay as WHERE. Equality
nested under `OrQuery` / `NotQuery` is left as WHERE.
- Login-name equals is a UNION arm that reuses `login_names3_users` SQL,
so it works inside OR as well as at the top level.
- EQUALS seeks use `LOWER(col)` for the index, then recheck `col =
value` so case-sensitive searches do not match extra rows.
EQUALS_IGNORE_CASE is unchanged.
- Empty OR arms are not skipped (that would change OR meaning); mixed
empty-OR falls back to WHERE. Login UI already omits empty arms.

### Indexes

- Setup 76: `CREATE INDEX CONCURRENTLY IF NOT EXISTS` on
`users14_username_lower_idx` `(instance_id, LOWER(username))` and
`users14_humans_phone_lower_idx` `(instance_id, LOWER(phone))`. Not
wrapped in a transaction.
- Before create, drop any leftover `pg_index.indisvalid = false` indexes
of those names (failed/cancelled `CREATE INDEX CONCURRENTLY`). `IF NOT
EXISTS` would otherwise skip rebuild.
- `Init()` creates the same indexes without `CONCURRENTLY` (empty
tables). Existing `users14_username` and email lower indexes are
unchanged.

# Additional Changes

- Unit tests assert UNION SQL for username OR email, email OR phone,
login name OR email, username OR username, nested `AndQuery(loginName,
org)`, and nested `AndQuery(Or(...), org)`; that contains / display-name
/ mixed OR / nested `OrQuery(AndQuery(...), ...)` do not UNION; that
email EQUALS rechecks the raw column; and that empty-OR is not
extracted.
- Setup 76 sqlmock coverage: missing `users14` skips work; valid/absent
indexes only create; invalid leftover username index is dropped then
both creates run.
- `TestServer_ListUsers_UsernameAndOr` checks live results: org AND
(alice OR bob) returns both users, sibling username AND is empty, nested
`AndQuery(Or(...), org)` returns both.

# Additional Context

- Base branch: `main`
- Extends the login-name equals rewrite from #12460 to
username/email/phone OR trees used by Login V2 search.

<!-- CURSOR_AGENT_PR_BODY_END -->

<div><a
href="https://cursor.com/agents/bc-1f043b1c-6168-45ca-9e9e-c18451994207?cursor_ref=pr_footer&cursor_cta=open_in_web"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://cursor.com/assets/images/open-in-web-dark.png"><source
media="(prefers-color-scheme: light)"
srcset="https://cursor.com/assets/images/open-in-web-light.png"><img
alt="Open in Web" width="114" height="28"
src="https://cursor.com/assets/images/open-in-web-dark.png"></picture></a>&nbsp;<a
href="https://cursor.com/background-agent?bcId=bc-1f043b1c-6168-45ca-9e9e-c18451994207&cursor_ref=pr_footer&cursor_cta=open_in_cursor"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://cursor.com/assets/images/open-in-cursor-dark.png"><source
media="(prefers-color-scheme: light)"
srcset="https://cursor.com/assets/images/open-in-cursor-light.png"><img
alt="Open in Cursor" width="131" height="28"
src="https://cursor.com/assets/images/open-in-cursor-dark.png"></picture></a>&nbsp;</div>

---------

Co-authored-by: Cursor Agent <[email protected]>
Co-authored-by: Silvan <[email protected]>
2026-09-08 13:11:16 +00:00

675 lines
24 KiB
Go

package setup
import (
"context"
"embed"
_ "embed"
"errors"
"fmt"
"net/http"
"os"
"os/signal"
"path"
"syscall"
"time"
"github.com/jackc/pgx/v5/pgconn"
"github.com/spf13/cobra"
"github.com/spf13/viper"
"github.com/zitadel/zitadel/backend/v3/instrumentation/logging"
"github.com/zitadel/zitadel/cmd/build"
"github.com/zitadel/zitadel/cmd/encryption"
"github.com/zitadel/zitadel/cmd/key"
"github.com/zitadel/zitadel/cmd/tls"
admin_handler "github.com/zitadel/zitadel/internal/admin/repository/eventsourcing/handler"
admin_view "github.com/zitadel/zitadel/internal/admin/repository/eventsourcing/view"
internal_authz "github.com/zitadel/zitadel/internal/api/authz"
auth_handler "github.com/zitadel/zitadel/internal/auth/repository/eventsourcing/handler"
auth_view "github.com/zitadel/zitadel/internal/auth/repository/eventsourcing/view"
"github.com/zitadel/zitadel/internal/authz"
authz_es "github.com/zitadel/zitadel/internal/authz/repository/eventsourcing/eventstore"
"github.com/zitadel/zitadel/internal/cache/connector"
"github.com/zitadel/zitadel/internal/command"
cryptoDB "github.com/zitadel/zitadel/internal/crypto/database"
"github.com/zitadel/zitadel/internal/database"
"github.com/zitadel/zitadel/internal/domain"
"github.com/zitadel/zitadel/internal/eventstore"
old_es "github.com/zitadel/zitadel/internal/eventstore/repository/sql"
new_es "github.com/zitadel/zitadel/internal/eventstore/v3"
"github.com/zitadel/zitadel/internal/i18n"
"github.com/zitadel/zitadel/internal/migration"
notify_handler "github.com/zitadel/zitadel/internal/notification"
"github.com/zitadel/zitadel/internal/query"
"github.com/zitadel/zitadel/internal/query/projection"
"github.com/zitadel/zitadel/internal/queue"
es_v4 "github.com/zitadel/zitadel/internal/v2/eventstore"
es_v4_pg "github.com/zitadel/zitadel/internal/v2/eventstore/postgres"
"github.com/zitadel/zitadel/internal/webauthn"
)
var (
//go:embed steps.yaml
defaultSteps []byte
stepFiles []string
)
func New() *cobra.Command {
cmd := &cobra.Command{
Use: "setup",
Short: "setup ZITADEL instance",
Long: `sets up data to start ZITADEL.
Requirements:
- postgreSQL`,
RunE: func(cmd *cobra.Command, args []string) (err error) {
defer func() {
logging.OnError(cmd.Context(), err).Error("zitadel setup command failed")
}()
err = tls.ModeFromFlag(cmd)
if err != nil {
return fmt.Errorf("invalid tlsMode: %w", err)
}
err = BindInitProjections(cmd)
if err != nil {
return fmt.Errorf("unable to bind \"init-projections\" flag: %w", err)
}
err = bindForMirror(cmd)
if err != nil {
return fmt.Errorf("unable to bind \"for-mirror\" flag: %w", err)
}
config, shutdown, err := NewConfig(cmd, viper.GetViper())
if err != nil {
return err
}
defer func() {
err = errors.Join(err, shutdown(cmd.Context()))
}()
steps, err := NewSteps(cmd.Context(), viper.New())
if err != nil {
return err
}
masterKey, err := key.MasterKey(cmd)
if err != nil {
return fmt.Errorf("no master key provided: %w", err)
}
return Setup(cmd.Context(), config, steps, masterKey)
},
}
cmd.AddCommand(NewCleanup())
Flags(cmd)
return cmd
}
func Flags(cmd *cobra.Command) {
cmd.PersistentFlags().StringArrayVar(&stepFiles, "steps", nil, "paths to step files to overwrite default steps")
cmd.Flags().Bool("init-projections", viper.GetBool("InitProjections"), "beta feature: initializes projections after they are created, allows smooth start as projections are up to date")
cmd.Flags().Bool("for-mirror", viper.GetBool("ForMirror"), "use this flag if you want to mirror your existing data")
key.AddMasterKeyFlag(cmd)
tls.AddTLSModeFlag(cmd)
}
func BindInitProjections(cmd *cobra.Command) error {
return viper.BindPFlag("InitProjections.Enabled", cmd.Flags().Lookup("init-projections"))
}
func bindForMirror(cmd *cobra.Command) error {
return viper.BindPFlag("ForMirror", cmd.Flags().Lookup("for-mirror"))
}
func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string) (err error) {
logging.Info(ctx, "setup started")
var setupErr error
ctx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM)
defer func() {
stop()
if setupErr == nil {
logging.Info(ctx, "setup completed")
return
}
if !errors.Is(setupErr, context.Canceled) {
// If Setup failed for some other reason than the context being cancelled,
// then this could be a fatal error we should not retry
logging.OnError(ctx, setupErr).Fatal("setup failed, skipping cleanup")
}
// if we're in the middle of long-running setup, run cleanup before exiting
// so if/when we're restarted we can pick up where we left off rather than
// booting into a broken state that requires manual intervention
// kubernetes will typically kill the pod after 30 seconds if the container does not exit
cleanupCtx, cleanupCancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
defer cleanupCancel()
err = Cleanup(cleanupCtx, config)
logging.OnError(ctx, err).Error("setup cleanup failed")
}()
i18n.MustLoadSupportedLanguagesFromDir()
dbClient, err := database.Connect(config.Database, false)
logging.OnError(ctx, err).Fatal("unable to connect to database")
config.Eventstore.Querier = old_es.NewPostgres(dbClient)
esV3 := new_es.NewEventstore(dbClient)
config.Eventstore.Pusher = esV3
config.Eventstore.Searcher = esV3
eventstoreClient := eventstore.NewEventstore(config.Eventstore)
logging.OnError(ctx, err).Fatal("unable to start eventstore")
eventstoreV4 := es_v4.NewEventstoreFromOne(es_v4_pg.New(dbClient, &es_v4_pg.Config{
MaxRetries: config.Eventstore.MaxRetries,
}))
httpClient := config.HTTPClient.NewClient()
steps.s1ProjectionTable = &ProjectionTable{dbClient: dbClient.DB}
steps.s2AssetsTable = &AssetTable{dbClient: dbClient.DB}
steps.FirstInstance.Skip = config.ForMirror || steps.FirstInstance.Skip
steps.FirstInstance.instanceSetup = config.DefaultInstance
steps.FirstInstance.userEncryptionKey = config.EncryptionKeys.User
steps.FirstInstance.smtpEncryptionKey = config.EncryptionKeys.SMTP
steps.FirstInstance.oidcEncryptionKey = config.EncryptionKeys.OIDC
steps.FirstInstance.masterKey = masterKey
steps.FirstInstance.db = dbClient
steps.FirstInstance.es = eventstoreClient
steps.FirstInstance.defaults = config.SystemDefaults
steps.FirstInstance.zitadelRoles = config.InternalAuthZ.RolePermissionMappings
steps.FirstInstance.externalDomain = config.ExternalDomain
steps.FirstInstance.externalSecure = config.ExternalSecure
steps.FirstInstance.externalPort = config.ExternalPort
steps.FirstInstance.defaultPaths = config.Login.DefaultPaths
steps.FirstInstance.httpClient = httpClient
steps.FirstInstance.denylist = config.HTTPClient.DenyList
steps.s5LastFailed = &LastFailed{dbClient: dbClient.DB}
steps.s6OwnerRemoveColumns = &OwnerRemoveColumns{dbClient: dbClient.DB}
steps.s7LogstoreTables = &LogstoreTables{dbClient: dbClient.DB, username: config.Database.Username()}
steps.s8AuthTokens = &AuthTokenIndexes{dbClient: dbClient}
steps.CorrectCreationDate.dbClient = dbClient
steps.s12AddOTPColumns = &AddOTPColumns{dbClient: dbClient}
steps.s13FixQuotaProjection = &FixQuotaConstraints{dbClient: dbClient}
steps.s14NewEventsTable = &NewEventsTable{dbClient: dbClient}
steps.s15CurrentStates = &CurrentProjectionState{dbClient: dbClient}
steps.s16UniqueConstraintsLower = &UniqueConstraintToLower{dbClient: dbClient}
steps.s17AddOffsetToUniqueConstraints = &AddOffsetToCurrentStates{dbClient: dbClient}
steps.s18AddLowerFieldsToLoginNames = &AddLowerFieldsToLoginNames{dbClient: dbClient}
steps.s19AddCurrentStatesIndex = &AddCurrentSequencesIndex{dbClient: dbClient}
steps.s20AddByUserSessionIndex = &AddByUserIndexToSession{dbClient: dbClient}
steps.s21AddBlockFieldToLimits = &AddBlockFieldToLimits{dbClient: dbClient}
steps.s22ActiveInstancesIndex = &ActiveInstanceEvents{dbClient: dbClient}
steps.s23CorrectGlobalUniqueConstraints = &CorrectGlobalUniqueConstraints{dbClient: dbClient}
steps.s24AddActorToAuthTokens = &AddActorToAuthTokens{dbClient: dbClient}
steps.s25User11AddLowerFieldsToVerifiedEmail = &User11AddLowerFieldsToVerifiedEmail{dbClient: dbClient}
steps.s26AuthUsers3 = &AuthUsers3{dbClient: dbClient}
steps.s27IDPTemplate6SAMLNameIDFormat = &IDPTemplate6SAMLNameIDFormat{dbClient: dbClient}
steps.s28AddFieldTable = &AddFieldTable{dbClient: dbClient}
steps.s29FillFieldsForProjectGrant = &FillFieldsForProjectGrant{eventstore: eventstoreClient}
steps.s30FillFieldsForOrgDomainVerified = &FillFieldsForOrgDomainVerified{eventstore: eventstoreClient}
steps.s31AddAggregateIndexToFields = &AddAggregateIndexToFields{dbClient: dbClient}
steps.s32AddAuthSessionID = &AddAuthSessionID{dbClient: dbClient}
steps.s33SMSConfigs3TwilioAddVerifyServiceSid = &SMSConfigs3TwilioAddVerifyServiceSid{dbClient: dbClient}
steps.s34AddCacheSchema = &AddCacheSchema{dbClient: dbClient}
steps.s35AddPositionToIndexEsWm = &AddPositionToIndexEsWm{dbClient: dbClient}
steps.s36FillV2Milestones = &FillV3Milestones{dbClient: dbClient, eventstore: eventstoreClient}
steps.s37Apps7OIDConfigsBackChannelLogoutURI = &Apps7OIDConfigsBackChannelLogoutURI{dbClient: dbClient}
steps.s38BackChannelLogoutNotificationStart = &BackChannelLogoutNotificationStart{dbClient: dbClient}
steps.s40InitPushFunc = &InitPushFunc{dbClient: dbClient}
steps.s42Apps7OIDCConfigsLoginVersion = &Apps7OIDCConfigsLoginVersion{dbClient: dbClient}
steps.s43CreateFieldsDomainIndex = &CreateFieldsDomainIndex{dbClient: dbClient}
steps.s44ReplaceCurrentSequencesIndex = &ReplaceCurrentSequencesIndex{dbClient: dbClient}
steps.s45CorrectProjectOwners = &CorrectProjectOwners{eventstore: eventstoreClient}
steps.s46InitPermissionFunctions = &InitPermissionFunctions{eventstoreClient: dbClient}
steps.s47FillMembershipFields = &FillMembershipFields{eventstore: eventstoreClient}
steps.s48Apps7SAMLConfigsLoginVersion = &Apps7SAMLConfigsLoginVersion{dbClient: dbClient}
steps.s49InitPermittedOrgsFunction = &InitPermittedOrgsFunction{eventstoreClient: dbClient}
steps.s50IDPTemplate6UsePKCE = &IDPTemplate6UsePKCE{dbClient: dbClient}
steps.s51IDPTemplate6RootCA = &IDPTemplate6RootCA{dbClient: dbClient}
steps.s52IDPTemplate6LDAP2 = &IDPTemplate6LDAP2{dbClient: dbClient}
steps.s53InitPermittedOrgsFunction = &InitPermittedOrgsFunction53{dbClient: dbClient}
steps.s54InstancePositionIndex = &InstancePositionIndex{dbClient: dbClient}
steps.s55ExecutionHandlerStart = &ExecutionHandlerStart{dbClient: dbClient}
steps.s56IDPTemplate6SAMLFederatedLogout = &IDPTemplate6SAMLFederatedLogout{dbClient: dbClient}
steps.s57CreateResourceCounts = &CreateResourceCounts{dbClient: dbClient}
steps.s58ReplaceLoginNames3View = &ReplaceLoginNames3View{dbClient: dbClient}
steps.s60GenerateSystemID = &GenerateSystemID{eventstore: eventstoreClient}
steps.s61IDPTemplate6SAMLSignatureAlgorithm = &IDPTemplate6SAMLSignatureAlgorithm{dbClient: dbClient}
steps.s62HTTPProviderAddSigningKey = &HTTPProviderAddSigningKey{dbClient: dbClient}
steps.s63AlterResourceCounts = &AlterResourceCounts{dbClient: dbClient}
steps.s64ChangePushPosition = &ChangePushPosition{dbClient: dbClient}
steps.s65FixUserMetadata5Index = &FixUserMetadata5Index{dbClient: dbClient}
steps.s66SessionRecoveryCodeCheckedAt = &SessionRecoveryCodeCheckedAt{dbClient: dbClient}
steps.s67SyncMemberRoleFields = &SyncMemberRoleFields{dbClient: dbClient}
steps.s68TargetAddPayloadTypeColumn = &TargetAddPayloadTypeColumn{dbClient: dbClient}
steps.s69CacheTablesLogged = &CacheTablesLogged{dbClient: dbClient}
steps.s70AddEventStoreCommandEnforceOwner = &AddEventStoreCommandEnforceOwnerColumn{dbClient: dbClient}
steps.s71JWTProvideAddAudienceColumn = &JWTProvideAddAudienceColumn{dbClient: dbClient}
steps.s72AddColumnsToLoginNamesView = &AddColumnsToLoginNamesView{dbClient: dbClient}
steps.s73FixUserGrantRoles = &FixUserGrantRoles{eventstore: eventstoreClient}
steps.s74Apps7OIDCConfigsAddRegistrationToken = &Apps7OIDCConfigsAddRegistrationToken{dbClient: dbClient}
steps.s75Apps7OIDCConfigsAddAppLinkConfig = &Apps7OIDCConfigsAddAppLinkConfig{dbClient: dbClient}
steps.s76Users14LoginEqualityIndexes = &Users14LoginEqualityIndexes{dbClient: dbClient}
err = projection.Create(ctx, dbClient, eventstoreClient, config.Projections, nil, nil, nil)
if err != nil {
return fmt.Errorf("unable to create projections: %w", err)
}
for _, step := range []migration.Migration{
steps.s14NewEventsTable,
steps.s40InitPushFunc,
steps.s1ProjectionTable,
steps.s2AssetsTable,
steps.s28AddFieldTable,
steps.s31AddAggregateIndexToFields,
steps.s46InitPermissionFunctions,
steps.FirstInstance,
steps.s5LastFailed,
steps.s6OwnerRemoveColumns,
steps.s7LogstoreTables,
steps.s8AuthTokens,
steps.s12AddOTPColumns,
steps.s13FixQuotaProjection,
steps.s15CurrentStates,
steps.s16UniqueConstraintsLower,
steps.s17AddOffsetToUniqueConstraints,
steps.s19AddCurrentStatesIndex,
steps.s20AddByUserSessionIndex,
steps.s22ActiveInstancesIndex,
steps.s23CorrectGlobalUniqueConstraints,
steps.s24AddActorToAuthTokens,
steps.s26AuthUsers3,
steps.s29FillFieldsForProjectGrant,
steps.s30FillFieldsForOrgDomainVerified,
steps.s34AddCacheSchema,
steps.s35AddPositionToIndexEsWm,
steps.s36FillV2Milestones,
steps.s38BackChannelLogoutNotificationStart,
steps.s44ReplaceCurrentSequencesIndex,
steps.s45CorrectProjectOwners,
steps.s47FillMembershipFields,
steps.s49InitPermittedOrgsFunction,
steps.s50IDPTemplate6UsePKCE,
steps.s51IDPTemplate6RootCA,
steps.s52IDPTemplate6LDAP2,
steps.s53InitPermittedOrgsFunction,
steps.s54InstancePositionIndex,
steps.s55ExecutionHandlerStart,
steps.s56IDPTemplate6SAMLFederatedLogout,
steps.s57CreateResourceCounts,
steps.s58ReplaceLoginNames3View,
steps.s60GenerateSystemID,
steps.s61IDPTemplate6SAMLSignatureAlgorithm,
steps.s62HTTPProviderAddSigningKey,
steps.s63AlterResourceCounts,
steps.s64ChangePushPosition,
steps.s65FixUserMetadata5Index,
steps.s67SyncMemberRoleFields,
steps.s69CacheTablesLogged,
steps.s70AddEventStoreCommandEnforceOwner,
steps.s76Users14LoginEqualityIndexes,
} {
setupErr = executeMigration(ctx, eventstoreClient, step, "migration failed")
if setupErr != nil {
return
}
}
commands, _, _, _ := startCommandsQueries(ctx, eventstoreClient, eventstoreV4, dbClient, masterKey, config, httpClient)
steps.s59SetupWebkeys = &SetupWebkeys{eventstore: eventstoreClient, commands: commands}
repeatableSteps := []migration.RepeatableMigration{
&externalConfigChange{
es: eventstoreClient,
ExternalDomain: config.ExternalDomain,
ExternalPort: config.ExternalPort,
ExternalSecure: config.ExternalSecure,
defaults: config.SystemDefaults,
},
&TransactionalTables{
dbClient: dbClient,
ShouldRecreateSchema: steps.RelationalTables.ShouldRecreateSchema,
},
&projectionTables{
es: eventstoreClient,
Version: build.Version(),
},
&DeleteStaleOrgFields{
eventstore: eventstoreClient,
},
&FillFieldsForInstanceDomains{
eventstore: eventstoreClient,
},
&SyncRolePermissions{
commands: commands,
eventstore: eventstoreClient,
rolePermissionMappings: config.InternalAuthZ.RolePermissionMappings,
},
&RiverMigrateRepeatable{
client: dbClient,
},
&eventstoreAutovacuum{
dbClient: dbClient,
Enabled: config.Eventstore.Autovacuum.Enabled,
VacuumThreshold: config.Eventstore.Autovacuum.VacuumThreshold,
AnalyzeThreshold: config.Eventstore.Autovacuum.AnalyzeThreshold,
},
}
repeatableSteps = append(repeatableSteps, triggerSteps(dbClient)...)
for _, repeatableStep := range repeatableSteps {
setupErr = executeMigration(ctx, eventstoreClient, repeatableStep, "unable to migrate repeatable step")
if setupErr != nil {
return
}
}
// These steps are executed after the repeatable steps because they add fields projections
for _, step := range []migration.Migration{
steps.s18AddLowerFieldsToLoginNames,
steps.s21AddBlockFieldToLimits,
steps.s25User11AddLowerFieldsToVerifiedEmail,
steps.s27IDPTemplate6SAMLNameIDFormat,
steps.s32AddAuthSessionID,
steps.s33SMSConfigs3TwilioAddVerifyServiceSid,
steps.s37Apps7OIDConfigsBackChannelLogoutURI,
steps.s42Apps7OIDCConfigsLoginVersion,
steps.s43CreateFieldsDomainIndex,
steps.s48Apps7SAMLConfigsLoginVersion,
steps.s59SetupWebkeys, // this step needs commands.
steps.s66SessionRecoveryCodeCheckedAt,
steps.s68TargetAddPayloadTypeColumn,
steps.s71JWTProvideAddAudienceColumn,
steps.s72AddColumnsToLoginNamesView,
steps.s73FixUserGrantRoles,
steps.s74Apps7OIDCConfigsAddRegistrationToken,
steps.s75Apps7OIDCConfigsAddAppLinkConfig,
} {
setupErr = executeMigration(ctx, eventstoreClient, step, "migration failed")
if setupErr != nil {
return
}
}
// projection initialization must be done last, since the steps above might add required columns to the projections
if !config.ForMirror && config.InitProjections.Enabled {
setupErr = initProjections(ctx, eventstoreClient)
if setupErr != nil {
return
}
}
return nil
}
func executeMigration(ctx context.Context, eventstoreClient *eventstore.Eventstore, step migration.Migration, errorMsg string) error {
err := migration.Migrate(ctx, eventstoreClient, step)
if err == nil {
return nil
}
logFields := []any{
"name", step.String(),
}
pgErr := new(pgconn.PgError)
if errors.As(err, &pgErr) {
logFields = append(logFields,
"severity", pgErr.Severity,
"code", pgErr.Code,
"message", pgErr.Message,
"detail", pgErr.Detail,
"hint", pgErr.Hint,
)
}
logging.WithError(ctx, err).Error(errorMsg, logFields...)
return err
}
// readStmt reads a single file from the embedded FS,
// under the folder/typ/filename path.
// Typ describes the database dialect and may be omitted if no
// dialect specific migration is specified.
func readStmt(fs embed.FS, folder, filename string) (string, error) {
stmt, err := fs.ReadFile(path.Join(folder, filename))
return string(stmt), err
}
type statement struct {
file string
query string
}
// readStatements reads all files from the embedded FS,
// under the folder/type path.
// Typ describes the database dialect and may be omitted if no
// dialect specific migration is specified.
func readStatements(fs embed.FS, folder string) ([]statement, error) {
dir, err := fs.ReadDir(folder)
if err != nil {
return nil, err
}
statements := make([]statement, len(dir))
for i, file := range dir {
statements[i].file = file.Name()
statements[i].query, err = readStmt(fs, folder, file.Name())
if err != nil {
return nil, err
}
}
return statements, nil
}
func startCommandsQueries(
ctx context.Context,
eventstoreClient *eventstore.Eventstore,
eventstoreV4 *es_v4.EventStore,
dbClient *database.DB,
masterKey string,
config *Config,
httpClient *http.Client,
) (
*command.Commands,
*query.Queries,
*admin_view.View,
*auth_view.View,
) {
keyStorage, err := cryptoDB.NewKeyStorage(dbClient, masterKey)
logging.OnError(ctx, err).Fatal("unable to start key storage")
keys, err := encryption.EnsureEncryptionKeys(ctx, config.EncryptionKeys, keyStorage)
logging.OnError(ctx, err).Fatal("unable to ensure encryption keys")
err = projection.Create(
ctx,
dbClient,
eventstoreClient,
projection.Config{
RetryFailedAfter: config.InitProjections.RetryFailedAfter,
MaxFailureCount: config.InitProjections.MaxFailureCount,
BulkLimit: config.InitProjections.BulkLimit,
},
keys.OIDC,
keys.SAML,
config.SystemAPIUsers,
)
logging.OnError(ctx, err).Fatal("unable to start projections")
staticStorage, err := config.AssetStorage.NewStorage(dbClient.DB)
logging.OnError(ctx, err).Fatal("unable to start asset storage")
adminView, err := admin_view.StartView(dbClient)
logging.OnError(ctx, err).Fatal("unable to start admin view")
admin_handler.Register(ctx,
admin_handler.Config{
Client: dbClient,
Eventstore: eventstoreClient,
BulkLimit: config.InitProjections.BulkLimit,
FailureCountUntilSkip: uint64(config.InitProjections.MaxFailureCount),
},
adminView,
staticStorage,
)
sessionTokenVerifier := internal_authz.SessionTokenVerifier(keys.OIDC)
cacheConnectors, err := connector.StartConnectors(config.Caches, dbClient)
logging.OnError(ctx, err).Fatal("unable to start caches")
queries, err := query.StartQueries(
ctx,
eventstoreClient,
eventstoreV4.Querier,
dbClient,
dbClient,
cacheConnectors,
config.Projections,
config.SystemDefaults,
keys.IDPConfig,
keys.OTP,
keys.OIDC,
keys.SAML,
keys.Target,
keys.SMS,
keys.SMTP,
config.InternalAuthZ.RolePermissionMappings,
sessionTokenVerifier,
func(q *query.Queries) domain.PermissionCheck {
return func(ctx context.Context, permission, orgID, resourceID string) (err error) {
return internal_authz.CheckPermission(ctx, &authz_es.UserMembershipRepo{Queries: q}, config.SystemAuthZ.RolePermissionMappings, config.InternalAuthZ.RolePermissionMappings, permission, orgID, resourceID)
}
},
0, // not needed for projections
nil, // not needed for projections
false,
config.DefaultInstance.SecretGenerators.ToMap(),
)
logging.OnError(ctx, err).Fatal("unable to start queries")
authView, err := auth_view.StartView(dbClient, keys.OIDC, queries, eventstoreClient)
logging.OnError(ctx, err).Fatal("unable to start auth view")
auth_handler.Register(ctx,
auth_handler.Config{
Client: dbClient,
Eventstore: eventstoreClient,
BulkLimit: config.InitProjections.BulkLimit,
FailureCountUntilSkip: uint64(config.InitProjections.MaxFailureCount),
},
authView,
queries,
)
authZRepo, err := authz.Start(queries, eventstoreClient, dbClient, keys.OIDC, config.ExternalSecure)
logging.OnError(ctx, err).Fatal("unable to start authz repo")
permissionCheck := func(ctx context.Context, permission, orgID, resourceID string) (err error) {
return internal_authz.CheckPermission(ctx, authZRepo, config.SystemAuthZ.RolePermissionMappings, config.InternalAuthZ.RolePermissionMappings, permission, orgID, resourceID)
}
commands, err := command.StartCommands(ctx,
eventstoreClient,
cacheConnectors,
config.SystemDefaults,
config.InternalAuthZ.RolePermissionMappings,
staticStorage,
&webauthn.Config{
DisplayName: config.WebAuthNName,
ExternalSecure: config.ExternalSecure,
},
config.ExternalDomain,
config.ExternalSecure,
config.ExternalPort,
keys.IDPConfig,
keys.OTP,
keys.SMTP,
keys.SMS,
keys.User,
keys.DomainVerification,
keys.SAML,
keys.Target,
keys.OIDC,
httpClient,
permissionCheck,
sessionTokenVerifier,
config.OIDC.DefaultAccessTokenLifetime,
config.OIDC.DefaultRefreshTokenExpiration,
config.OIDC.DefaultRefreshTokenIdleExpiration,
config.DefaultInstance.SecretGenerators,
config.Login.DefaultPaths,
config.HTTPClient.DenyList,
)
logging.OnError(ctx, err).Fatal("unable to start commands")
q, err := queue.NewQueue(&queue.Config{
Client: dbClient,
})
logging.OnError(ctx, err).Fatal("unable to init queue")
notify_handler.Register(
ctx,
config.Projections.Customizations["notifications"],
config.Projections.Customizations["notificationsquotas"],
config.Projections.Customizations["backchannel"],
config.Projections.Customizations["telemetry"],
config.Notifications,
config.OIDC.BackChannelLogoutConfig(),
*config.Telemetry,
config.ExternalDomain,
config.ExternalPort,
config.ExternalSecure,
commands,
queries,
eventstoreClient,
config.Login.DefaultPaths.DefaultOTPEmailURLTemplate,
config.SystemDefaults.Notifications.FileSystemPath,
keys.User,
keys.SMTP,
keys.SMS,
q,
httpClient,
)
return commands, queries, adminView, authView
}
func initProjections(
ctx context.Context,
eventstoreClient *eventstore.Eventstore,
) error {
for _, p := range projection.Projections() {
if err := migration.Migrate(ctx, eventstoreClient, p); err != nil {
logging.WithError(ctx, err).Error("projection migration failed", "name", p.String())
return err
}
}
for _, p := range admin_handler.Projections() {
if err := migration.Migrate(ctx, eventstoreClient, p); err != nil {
logging.WithError(ctx, err).Error("admin schema migration failed", "name", p.String())
return err
}
}
for _, p := range auth_handler.Projections() {
if err := migration.Migrate(ctx, eventstoreClient, p); err != nil {
logging.WithError(ctx, err).Error("auth schema migration failed", "name", p.String())
return err
}
}
for _, p := range notify_handler.Projections() {
if err := migration.Migrate(ctx, eventstoreClient, p); err != nil {
logging.WithError(ctx, err).Error("notification migration failed", "name", p.String())
return err
}
}
return nil
}