feat(hydra): split up persister

GitOrigin-RevId: 203cf926c1613fcbb20393c5b7d0af25c7aecb15

Author: zepatrik

.reports/dep-licenses.csv

@@ -3,5 +3,4 @@
 "github.com/ory/x","Apache-2.0"
 "github.com/stretchr/testify","MIT"
 "go.opentelemetry.io/otel/sdk","Apache-2.0"
-"go.opentelemetry.io/otel/sdk","BSD-3-Clause"
 

aead/aesgcm.go

@@ -62,7 +62,7 @@ func (c *AESGCM) Decrypt(ctx context.Context, ciphertext string, aad []byte) (pl
 	return nil, err
 }
 
-func (c *AESGCM) decrypt(ciphertext []byte, key, additionalData []byte) ([]byte, error) {
+func (*AESGCM) decrypt(ciphertext, key, additionalData []byte) ([]byte, error) {
 	if len(key) != 32 {
 		return nil, errors.Errorf("key must be exactly 32 long bytes, got %d bytes", len(key))
 	}

cmd/cli/handler_migrate.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/ory/hydra/v2/driver"
 	"github.com/ory/hydra/v2/driver/config"
-	"github.com/ory/hydra/v2/persistence"
+	"github.com/ory/hydra/v2/persistence/sql"
 	"github.com/ory/x/cmdx"
 	"github.com/ory/x/configx"
 	"github.com/ory/x/popx"
@@ -26,7 +26,7 @@ func newMigrateHandler(dOpts []driver.OptionsModifier) *MigrateHandler {
 	}
 }
 
-func (h *MigrateHandler) makePersister(cmd *cobra.Command, args []string) (p persistence.Persister, err error) {
+func (h *MigrateHandler) makeMigrationManager(cmd *cobra.Command, args []string) (*sql.MigrationManager, error) {
 	opts := append([]driver.OptionsModifier{
 		driver.WithConfigOptions(
 			configx.SkipValidation(),
@@ -52,27 +52,27 @@ func (h *MigrateHandler) makePersister(cmd *cobra.Command, args []string) (p per
 		return nil, cmdx.FailSilently(cmd)
 	}
 
-	return d.Persister(), nil
+	return d.Migrator(), nil
 }
 
 func (h *MigrateHandler) MigrateSQLUp(cmd *cobra.Command, args []string) (err error) {
-	p, err := h.makePersister(cmd, args)
+	p, err := h.makeMigrationManager(cmd, args)
 	if err != nil {
 		return err
 	}
 	return popx.MigrateSQLUp(cmd, p)
 }
 
 func (h *MigrateHandler) MigrateSQLDown(cmd *cobra.Command, args []string) (err error) {
-	p, err := h.makePersister(cmd, args)
+	p, err := h.makeMigrationManager(cmd, args)
 	if err != nil {
 		return err
 	}
 	return popx.MigrateSQLDown(cmd, p)
 }
 
 func (h *MigrateHandler) MigrateStatus(cmd *cobra.Command, args []string) error {
-	p, err := h.makePersister(cmd, args)
+	p, err := h.makeMigrationManager(cmd, args)
 	if err != nil {
 		return err
 	}

cmd/server/helper_cert.go

@@ -13,7 +13,6 @@ import (
 	"sync"
 
 	"github.com/go-jose/go-jose/v3"
-	"github.com/gofrs/uuid"
 	"github.com/pkg/errors"
 
 	"github.com/ory/hydra/v2/driver"
@@ -54,7 +53,7 @@ func GetOrCreateTLSCertificate(ctx context.Context, d *driver.RegistrySQL, tlsCo
 	}
 
 	// no certificates configured: self-sign a new cert
-	priv, err := jwk.GetOrGenerateKeys(ctx, d, d.SoftwareKeyManager(), TlsKeyName, uuid.Must(uuid.NewV4()).String(), "RS256")
+	priv, err := jwk.GetOrGenerateKeys(ctx, d, d.KeyManager(), TlsKeyName, "RS256")
 	if err != nil {
 		d.Logger().WithError(err).Fatal("Unable to fetch or generate HTTPS TLS key pair")
 		return nil // in case Fatal is hooked
@@ -68,12 +67,12 @@ func GetOrCreateTLSCertificate(ctx context.Context, d *driver.RegistrySQL, tlsCo
 		}
 
 		AttachCertificate(priv, cert)
-		if err := d.SoftwareKeyManager().DeleteKey(ctx, TlsKeyName, priv.KeyID); err != nil {
+		if err := d.KeyManager().DeleteKey(ctx, TlsKeyName, priv.KeyID); err != nil {
 			d.Logger().WithError(err).Fatal(`Could not update (delete) the self signed TLS certificate`)
 			return nil // in case Fatal is hooked
 		}
 
-		if err := d.SoftwareKeyManager().AddKey(ctx, TlsKeyName, priv); err != nil {
+		if err := d.KeyManager().AddKey(ctx, TlsKeyName, priv); err != nil {
 			d.Logger().WithError(err).Fatalf(`Could not update (add) the self signed TLS certificate: %s %x %d`, cert.SignatureAlgorithm, cert.Signature, len(cert.Signature))
 			return nil // in case Fatalf is hooked
 		}

driver/registry.go

@@ -25,8 +25,6 @@ import (
 )
 
 type registry interface {
-	dbal.Driver
-
 	x.HTTPClientProvider
 
 	contextx.Provider

driver/registry_sql.go

@@ -64,7 +64,6 @@ type RegistrySQL struct {
 	cv              *client.Validator
 	ctxer           contextx.Contextualizer
 	hh              *healthx.Handler
-	migrationStatus *popx.MigrationStatuses
 	kc              *aead.AESGCM
 	flowc           *aead.XChaCha20Poly1305
 	cos             consent.Strategy
@@ -76,7 +75,7 @@ type RegistrySQL struct {
 	trc             *otelx.Tracer
 	tracerWrapper   func(*otelx.Tracer) *otelx.Tracer
 	arhs            []oauth2.AccessRequestHook
-	persister       persistence.Persister
+	basePersister   *sql.BasePersister
 	oc              fosite.Configurator
 	oidcs           jwk.JWTSigner
 	ats             jwk.JWTSigner
@@ -87,10 +86,11 @@ type RegistrySQL struct {
 	publicCORS      *cors.Cors
 	kratos          kratos.Client
 	fositeFactories []fositex.Factory
+	migrator        *sql.MigrationManager
 
-	defaultKeyManager jwk.Manager
-	initialPing       func(r *RegistrySQL) error
-	middlewares       []negroni.Handler
+	keyManager  jwk.Manager
+	initialPing func(ctx context.Context, l *logrusx.Logger, p *sql.BasePersister) error
+	middlewares []negroni.Handler
 }
 
 var (
@@ -101,12 +101,10 @@ var (
 // defaultInitialPing is the default function that will be called within RegistrySQL.Init to make sure
 // the database is reachable. It can be injected for test purposes by changing the value
 // of RegistrySQL.initialPing.
-func defaultInitialPing(m *RegistrySQL) error {
-	if err := resilience.Retry(m.l, 5*time.Second, 5*time.Minute, m.Ping); err != nil {
-		m.Logger().Print("Could not ping database: ", err)
-		return errors.WithStack(err)
-	}
-	return nil
+func defaultInitialPing(ctx context.Context, l *logrusx.Logger, p *sql.BasePersister) error {
+	return errors.WithStack(resilience.Retry(l, 5*time.Second, 5*time.Minute, func() error {
+		return p.Ping(ctx)
+	}))
 }
 
 func (m *RegistrySQL) Init(
@@ -116,7 +114,7 @@ func (m *RegistrySQL) Init(
 	extraMigrations []fs.FS,
 	goMigrations []popx.Migration,
 ) error {
-	if m.persister == nil {
+	if m.basePersister == nil {
 		if m.Config().CGroupsV1AutoMaxProcsEnabled() {
 			_, err := maxprocs.Set(maxprocs.Logger(m.Logger().Infof))
 			if err != nil {
@@ -146,59 +144,40 @@ func (m *RegistrySQL) Init(
 			return errors.WithStack(err)
 		}
 
-		p, err := sql.NewPersister(c, m, m.Config(), extraMigrations, goMigrations)
-		if err != nil {
-			return err
-		}
-		m.persister = p
-		if err := m.initialPing(m); err != nil {
+		m.basePersister = sql.NewBasePersister(c, m)
+		if err := m.initialPing(ctx, m.Logger(), m.basePersister); err != nil {
+			m.Logger().Print("Could not ping database: ", err)
 			return err
 		}
 
-		if m.Config().HSMEnabled() {
-			hardwareKeyManager := hsm.NewKeyManager(m.HSMContext(), m.Config())
-			m.defaultKeyManager = jwk.NewManagerStrategy(hardwareKeyManager, m.persister)
-		} else {
-			m.defaultKeyManager = m.persister
-		}
+		m.migrator = sql.NewMigrationManager(c, m, extraMigrations, goMigrations)
 
 		// if dsn is memory we have to run the migrations on every start
 		// use case - such as
 		// - just in memory
 		// - shared connection
 		// - shared but unique in the same process
 		// see: https://sqlite.org/inmemorydb.html
-		if dbal.IsMemorySQLite(m.Config().DSN()) {
-			m.Logger().Print("Hydra is running migrations on every startup as DSN is memory.\n")
-			m.Logger().Print("This means your data is lost when Hydra terminates.\n")
-			if err := p.MigrateUp(context.Background()); err != nil {
-				return err
-			}
-		} else if migrate {
-			if err := p.MigrateUp(context.Background()); err != nil {
+		switch {
+		case dbal.IsMemorySQLite(m.Config().DSN()):
+			m.Logger().Println("Hydra is running migrations on every startup as DSN is memory.")
+			m.Logger().Println("This means your data is lost when Hydra terminates.")
+			fallthrough
+		case migrate:
+			if err := m.migrator.MigrateUp(ctx); err != nil {
 				return err
 			}
 		}
 
-		if skipNetworkInit {
-			m.persister = p
-		} else {
-			net, err := p.DetermineNetwork(ctx)
+		if !skipNetworkInit {
+			net, err := m.basePersister.DetermineNetwork(ctx)
 			if err != nil {
 				m.Logger().WithError(err).Warnf("Unable to determine network, retrying.")
 				return err
 			}
 
-			m.persister = p.WithFallbackNetworkID(net.ID)
-		}
-
-		if m.Config().HSMEnabled() {
-			hardwareKeyManager := hsm.NewKeyManager(m.HSMContext(), m.Config())
-			m.defaultKeyManager = jwk.NewManagerStrategy(hardwareKeyManager, m.persister)
-		} else {
-			m.defaultKeyManager = m.persister
+			m.basePersister = m.basePersister.WithFallbackNetworkID(net.ID)
 		}
-
 	}
 
 	return nil
@@ -211,11 +190,7 @@ func (m *RegistrySQL) alwaysCanHandle(dsn string) bool {
 }
 
 func (m *RegistrySQL) PingContext(ctx context.Context) error {
-	return m.Persister().Ping(ctx)
-}
-
-func (m *RegistrySQL) Ping() error {
-	return m.PingContext(context.Background())
+	return m.basePersister.Ping(ctx)
 }
 
 func (m *RegistrySQL) ClientManager() client.Manager {
@@ -231,11 +206,16 @@ func (m *RegistrySQL) OAuth2Storage() x.FositeStorer {
 }
 
 func (m *RegistrySQL) KeyManager() jwk.Manager {
-	return m.defaultKeyManager
-}
-
-func (m *RegistrySQL) SoftwareKeyManager() jwk.Manager {
-	return m.Persister()
+	if m.keyManager == nil {
+		softwareKeyManager := &sql.JWKPersister{BasePersister: m.basePersister}
+		if m.Config().HSMEnabled() {
+			hardwareKeyManager := hsm.NewKeyManager(m.HSMContext(), m.Config())
+			m.keyManager = jwk.NewManagerStrategy(hardwareKeyManager, softwareKeyManager)
+		} else {
+			m.keyManager = softwareKeyManager
+		}
+	}
+	return m.keyManager
 }
 
 func (m *RegistrySQL) GrantManager() trust.GrantManager {
@@ -330,11 +310,7 @@ func (m *RegistrySQL) HealthHandler() *healthx.Handler {
 				return m.PingContext(r.Context())
 			},
 			"migrations": func(r *http.Request) error {
-				if m.migrationStatus != nil && !m.migrationStatus.HasPending() {
-					return nil
-				}
-
-				status, err := m.Persister().MigrationStatus(r.Context())
+				status, err := m.migrator.MigrationStatus(r.Context())
 				if err != nil {
 					return err
 				}
@@ -344,8 +320,6 @@ func (m *RegistrySQL) HealthHandler() *healthx.Handler {
 					m.Logger().WithField("status", fmt.Sprintf("%+v", status)).WithError(err).Warn("Instance is not yet ready because migrations have not yet been fully applied.")
 					return err
 				}
-
-				m.migrationStatus = &status
 				return nil
 			},
 		})
@@ -529,7 +503,7 @@ func (m *RegistrySQL) OpenIDConnectRequestValidator() *openid.OpenIDConnectReque
 }
 
 func (m *RegistrySQL) Networker() x.Networker {
-	return m.persister
+	return m.basePersister
 }
 
 func (m *RegistrySQL) SubjectIdentifierAlgorithm(ctx context.Context) map[string]consent.SubjectIdentifierAlgorithm {
@@ -569,7 +543,7 @@ func (m *RegistrySQL) Tracer(_ context.Context) *otelx.Tracer {
 }
 
 func (m *RegistrySQL) Persister() persistence.Persister {
-	return m.persister
+	return sql.NewPersister(m.basePersister, m)
 }
 
 // Config returns the configuration for the given context. It may or may not be the same as the global configuration.
@@ -609,3 +583,7 @@ func (m *RegistrySQL) Kratos() kratos.Client {
 func (m *RegistrySQL) HTTPMiddlewares() []negroni.Handler {
 	return m.middlewares
 }
+
+func (m *RegistrySQL) Migrator() *sql.MigrationManager {
+	return m.migrator
+}

driver/registry_sql_test.go

@@ -64,7 +64,6 @@ func TestRegistrySQL_newKeyStrategy_handlesNetworkError(t *testing.T) {
 
 	l := logrusx.New("", "", logrusx.WithHook(&hook))
 	l.Logrus().SetOutput(io.Discard)
-	l.Logrus().ExitFunc = func(int) {} // Override the exit func to avoid call to os.Exit
 
 	// Create a config and set a valid but unresolvable DSN
 	c := config.MustNew(t, l,
@@ -88,7 +87,7 @@ func TestRegistrySQL_newKeyStrategy_handlesNetworkError(t *testing.T) {
 		"snizzles",
 	)
 
-	assert.Equal(t, logrus.FatalLevel, hook.LastEntry().Level)
+	assert.Equal(t, logrus.InfoLevel, hook.LastEntry().Level)
 	assert.Contains(t, hook.LastEntry().Message, "snizzles")
 }
 
@@ -158,8 +157,7 @@ func TestDefaultKeyManager_HsmDisabled(t *testing.T) {
 		}),
 	)
 	require.NoError(t, err)
-	assert.IsType(t, &sql.Persister{}, r.KeyManager())
-	assert.IsType(t, &sql.Persister{}, r.SoftwareKeyManager())
+	assert.IsType(t, &sql.JWKPersister{}, r.KeyManager())
 }
 
 func TestDbUnknownTableColumns(t *testing.T) {
@@ -199,14 +197,13 @@ func TestDbUnknownTableColumns(t *testing.T) {
 	})
 }
 
-func sussessfulPing(r *RegistrySQL) error {
+func sussessfulPing(context.Context, *logrusx.Logger, *sql.BasePersister) error {
 	// fake that ping is successful
 	return nil
 }
 
-func failedPing(err error) func(r *RegistrySQL) error {
-	return func(r *RegistrySQL) error {
-		r.Logger().Fatal(err.Error())
+func failedPing(err error) func(context.Context, *logrusx.Logger, *sql.BasePersister) error {
+	return func(context.Context, *logrusx.Logger, *sql.BasePersister) error {
 		return pkgerr.WithStack(err)
 	}
 }