Files
3ff4fbc7e4 fix(eventstore): add owners to unique constraints (#12740)
<!-- CURSOR_AGENT_PR_BODY_BEGIN -->
# Which Problems Are Solved

Bulk unique-constraint deletes (org, IDP with many links, user, project)
reconstruct every uniqueness key in Go and emit N `DELETE`s. That can
OOM on large tenants.

A one-shot owner backfill is also not enough: during a rolling deploy,
older binaries still insert `owners = '{}'`, so a later owner-only
delete would miss those rows.

Owner-only org/user/grant delete only matches rows whose `owners` bag
contains that tag. Several grant and member types stamped the aggregate
resource owner, which is not always the user’s home org or the granted
org. Username-scope rewrites also dropped `user:` tags, so owner-only
user delete could miss rewritten username rows.

Command must not switch to owner-only delete on the first 79 of a
version, because leftover `'{}'` rows from rolling-update overlap are
only backfilled on a later 79. A failed 79 must not turn owner-delete
on, and backfill joins must not stamp the wrong org (unverified domains,
org-scoped usernames colliding with instance-scoped ones).

# How the Problems Are Solved

Uniqueness stays the existing primary key `(instance_id, unique_type,
unique_field)`. `owners` is only a lifecycle bag of tags (`org:<id>`,
`user:<id>`, `idp:<id>`, `project:<id>`, `grant:<id>`).

- Add `owners TEXT[] NOT NULL DEFAULT '{}'` on
`eventstore.unique_constraints` (init SQL for greenfield, setup 78 for
existing clusters).
- Create `unique_constraints_owners_gin` with `CREATE INDEX
CONCURRENTLY` **outside a transaction** (same pattern as step 76).
Built-in `GIN (owners)` only; no `btree_gin`.
- Stamp `WithOwners` on **add** events. Field-level removes ignore
owners. `RemoveByOwner` stores the tag in `Owners` and both v1/v3 and v2
pushers run `DELETE … WHERE instance_id = $1 AND owners @>
ARRAY[$tag]::text[]`.
- Setup 79 backfills `owners` from projections via numbered
`cmd/setup/79/*.sql` files. Table names are interpolated from
`internal/query/projection` constants. Matches use `unique_field IN (x,
lower(x))` so the primary key can be used. Missing projection tables
(`42P01`) are skipped.
- Org domains are tagged only when `is_verified`. Instance-scoped
usernames join only users that are not org-scoped (organization settings
or domain policy `user_login_must_be_domain`). Every username row still
gets `org:` + `user:` of that user.
- 79 is a **repeatable** migration. `Check` re-runs when
`lastRun["version"]` changed or when setup-only `ForceFinalize` is set
and `lastRun["finalized"]` is still false. `Finalized` is written only
after Execute succeeds.
- Failed setup events keep the last successful `lastRun`. Command reads
`finalized` only from `system.migration.repeatable.done`.
- Org / user / project remove skip N-key event replay **when that
`finalized` bit is true**. Otherwise they keep today’s N-key lists
**and** still emit `RemoveByOwner`. The “ready” result is cached in
memory once true.

Setup writes `finalized` on the 79 repeatable.done payload:

- First 79 of a version: backfill only, `finalized=false` (unless
`ForceFinalize` is already set).
- Later 79: version change or
`BackfillUniqueConstraintOwners.ForceFinalize` /
`ZITADEL_BACKFILLUNIQUECONSTRAINTOWNERS_FORCEFINALIZE` after the rolling
update is done. That run backfills leftovers and writes
`finalized=true`.

Member and grant owner bags now include every org that should delete the
row:

- Instance, org, and project members also stamp
`org:<user_resource_owner>`.
- User grants stamp grant, user, project, and granted orgs.
- Project grant members stamp the project org, user org, and granted
org.
- Project grants stamp both orgs plus `grant:<id>`. Grant remove emits
`RemoveByOwner(grant)`.
- Username-scope rewrites keep `user:<id>` as well as `org:`.
- User/grant orgs are carried as unexported add-event fields (not
persisted JSON). `WithOwners` skips empty and duplicate tags.

# Additional Changes

- Table-driven owner-tag tests, including project grant (two `org:` tags
plus `grant:`), cross-org members, user grants, and username-scope
rewrite `user:` tags.
- Table-driven 79 `Check` + `finalizeAfterSuccess` tests for version
change, sequential version bumps, ForceFinalize restart, and
already-finalized skip.
- Command tests: a failed 79 event with `finalized=true` does not enable
owner-delete.
- Integration fixtures (verified vs unverified org domain;
instance-scoped `bob` vs org-scoped `bob`).
- Instance-only and global unique types stay untagged (`'{}'`). Instance
wipe still uses `UniqueConstraintInstanceRemove`.
- Pusher owner cases live in `eventstore_pusher_owners_test.go` so
`eventstore_pusher_test.go` stays under 1k lines.

# Additional Context

A later cleanup can delete the N-key replay helpers once a release that
ran 79 a second time (or an operator who set `ForceFinalize`) is in
production. That is not required for correctness.

<!-- CURSOR_AGENT_PR_BODY_END -->

<div><a
href="https://cursor.com/agents/bc-44f9bc9b-9a93-4cd9-8dd1-663a7b9aad2a?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-44f9bc9b-9a93-4cd9-8dd1-663a7b9aad2a&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-18 15:08:42 +02:00

685 lines
25 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}
steps.s77StampEventPositionAtInsert = &StampEventPositionAtInsert{dbClient: dbClient}
steps.s78UniqueConstraintOwners = &UniqueConstraintOwners{dbClient: dbClient}
if steps.BackfillUniqueConstraintOwners == nil {
steps.BackfillUniqueConstraintOwners = &BackfillUniqueConstraintOwners{}
}
steps.BackfillUniqueConstraintOwners.dbClient = dbClient
steps.BackfillUniqueConstraintOwners.Version = build.Version()
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,
steps.s77StampEventPositionAtInsert,
steps.s78UniqueConstraintOwners,
steps.BackfillUniqueConstraintOwners,
} {
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
}