upstream-ory-keto/oryx/popx/migrator.go

// Copyright © 2023 Ory Corp
// SPDX-License-Identifier: Apache-2.0

package popx

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"database/sql/driver"
	"encoding/hex"
	"fmt"
	"math"
	"os"
	"path/filepath"
	"regexp"
	"slices"
	"strings"
	"testing"
	"time"

	"github.com/cockroachdb/cockroach-go/v2/crdb"
	"github.com/gofrs/flock"
	"github.com/pkg/errors"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
	moderncsqlite "modernc.org/sqlite"

	"github.com/ory/pop/v6"
	"github.com/ory/x/cmdx"
	"github.com/ory/x/dbal"
	"github.com/ory/x/logrusx"
	"github.com/ory/x/otelx"
	"github.com/ory/x/sqlcon"
)

const (
	Pending          = "Pending"
	Applied          = "Applied"
	tracingComponent = "github.com/ory/x/popx"
)

func (mb *MigrationBox) shouldNotUseTransaction(m Migration) bool {
	return m.Autocommit || mb.noTxDDL()
}

// Up runs pending "up" migrations and applies them to the database.
func (mb *MigrationBox) Up(ctx context.Context) error {
	_, err := mb.UpTo(ctx, 0)
	return errors.WithStack(err)
}

// sqliteTemplatePath computes the path of the SQLite template for this
// migration set. The path is deterministic and content-addressed, so a changed
// migration set cannot reuse an older template.
func (mb *MigrationBox) sqliteTemplatePath() string {
	mfs := mb.migrationsUp.sortAndFilter(mb.c.Dialect.Name(), mb.migrationFallbacks()...)
	h := sha256.New()
	for _, mi := range mfs {
		for _, part := range []string{
			mi.Version,
			mi.Path,
			mi.Name,
			mi.Direction,
			mi.Type,
			mi.DBType,
			mi.Content,
		} {
			_, _ = h.Write([]byte(part))
			_, _ = h.Write([]byte{0})
		}
	}
	dir := mb.sqliteTemplateCacheDir
	if dir == "" {
		dir = os.TempDir()
	}
	return filepath.Join(dir, "ory-popx-sqlite-template-"+hex.EncodeToString(h.Sum(nil))+".sqlite")
}

// restoreSQLiteOnline streams the contents of srcPath into the database
// represented by db using SQLite's online backup API. The destination
// connection stays open throughout, so the *sql.DB and any pop.Connection
// holding it remain valid.
//
// pop wraps the modernc.org/sqlite driver with otelsql for tracing; we
// unwrap via the otelsql Raw() accessor before reaching the driver-specific
// NewRestore method on the underlying *sqlite.conn.
func restoreSQLiteOnline(ctx context.Context, db *sql.DB, srcPath string) error {
	conn, err := db.Conn(ctx)
	if err != nil {
		return errors.Wrap(err, "failed to acquire sql.Conn for restore")
	}
	defer func() { _ = conn.Close() }()

	return conn.Raw(func(driverConn any) error {
		if w, ok := driverConn.(interface{ Raw() driver.Conn }); ok {
			driverConn = w.Raw()
		}
		restorer, ok := driverConn.(interface {
			NewRestore(srcUri string) (*moderncsqlite.Backup, error)
		})
		if !ok {
			return errors.Errorf("driver %T does not support online restore", driverConn)
		}
		// See: https://sqlite.org/backup.html .
		backup, err := restorer.NewRestore(srcPath)
		if err != nil {
			return errors.Wrap(err, "NewRestore failed")
		}
		// Step(-1) copies all remaining pages in one call.
		for {
			more, stepErr := backup.Step(-1)
			if stepErr != nil {
				_ = backup.Finish()
				return errors.Wrap(stepErr, "backup.Step failed")
			}
			if !more {
				break
			}
		}
		return errors.Wrap(backup.Finish(), "backup.Finish failed")
	})
}

// UpTo runs up to step "up" migrations and applies them to the database.
// If step <= 0 all pending migrations are run.
func (mb *MigrationBox) UpTo(ctx context.Context, step int) (applied int, err error) {
	ctx, span := startSpan(ctx, MigrationUpOpName, trace.WithAttributes(attribute.Int("step", step)))
	defer otelx.End(span, &err)

	c := mb.c.WithContext(ctx)

	newDbFileName, isOnDiskSQLite := sqliteFilePath(mb.c.URL())
	isSQLite := mb.c.Dialect.Name() == "sqlite3"
	isOnDiskSQLite = isSQLite && isOnDiskSQLite

	// For test SQLite databases, try to restore a pre-migrated template
	// using SQLite's online backup API. The restore streams pages directly into
	// the open connection, so mb.c stays valid throughout and any holders of
	// it (including WithContext copies) keep working.
	if testing.Testing() && isOnDiskSQLite && step <= 0 && !mb.disableGoldenDatabase {
		templatePath := mb.sqliteTemplatePath()
		// Only restore onto a fresh (uninitialized) database. If the
		// migration table already exists, migrations were previously applied
		// — either directly or via an earlier template restore.
		mtn := sanitizedMigrationTableName(mb.c)
		var existingTableCount int
		alreadyMigrated := mb.c.Store.SQLDB().QueryRow(
			"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?", mtn,
		).Scan(&existingTableCount) == nil && existingTableCount > 0
		if !alreadyMigrated {
			if mkdirErr := os.MkdirAll(filepath.Dir(templatePath), 0o750); mkdirErr != nil {
				mb.l.Errorf("Failed to create SQLite template cache directory, applying migrations: path=%s err=%v", filepath.Dir(templatePath), mkdirErr)
			} else {
				// Test packages run as separate processes. Serialize the first
				// template build so only one of them pays the migration cost;
				// waiters restore the completed template after acquiring the
				// lock. The template itself is still installed atomically.
				lock := flock.New(templatePath + ".lock")
				if lockErr := lock.Lock(); lockErr != nil {
					mb.l.Errorf("Failed to lock SQLite template, applying migrations: path=%s err=%v", templatePath, lockErr)
				} else {
					defer func() {
						if unlockErr := lock.Unlock(); unlockErr != nil {
							mb.l.Errorf("Failed to unlock SQLite template: path=%s err=%v", templatePath, unlockErr)
						}
					}()
				}
			}

			if _, statErr := os.Stat(templatePath); statErr == nil {
				if restoreErr := restoreSQLiteOnline(ctx, mb.c.Store.SQLDB(), templatePath); restoreErr != nil {
					// Restore failed; log and fall through to the normal migration path.
					mb.l.Errorf("Failed to restore SQLite template, applying migrations: src=%s dst=%s err=%v", templatePath, newDbFileName, restoreErr)
				} else {
					mfs := mb.migrationsUp.sortAndFilter(mb.c.Dialect.Name(), mb.migrationFallbacks()...)
					mb.l.Infof("Skipped applying %d migrations using SQLite template: path=%s", len(mfs), templatePath)
					return len(mfs), nil
				}
			} else if !errors.Is(statErr, os.ErrNotExist) {
				mb.l.Errorf("Failed to inspect SQLite template, applying migrations: path=%s err=%v", templatePath, statErr)
			}
		}
	}

	// Keep applied across whole-run retry attempts. Every migration counted here
	// has already committed its schema_migration row, so it remains part of this
	// UpTo call's step budget even when a later migration retries.
	err = mb.exec(ctx, func() error {
		mtn := sanitizedMigrationTableName(c)
		mfs := mb.migrationsUp.sortAndFilter(mb.c.Dialect.Name(), mb.migrationFallbacks()...)
		for _, mi := range mfs {
			l := mb.l.WithField("version", mi.Version).WithField("migration_name", mi.Name).WithField("migration_file", mi.Path)

			appliedMigrations := make([]string, 0, 2)
			legacyVersion := mi.Version
			if len(legacyVersion) > 14 {
				legacyVersion = legacyVersion[:14]
			}
			err := c.RawQuery(fmt.Sprintf("SELECT version FROM %s WHERE version IN (?, ?)", mtn), mi.Version, legacyVersion).All(&appliedMigrations)
			if err != nil {
				return errors.Wrapf(err, "problem checking for migration version %s", mi.Version)
			}

			if slices.Contains(appliedMigrations, mi.Version) {
				l.Debug("Migration has already been applied, skipping.")
				continue
			}

			if slices.Contains(appliedMigrations, legacyVersion) {
				l.WithField("legacy_version", legacyVersion).WithField("migration_table", mtn).Debug("Migration has already been applied in a legacy migration run. Updating version in migration table.")
				if err := mb.isolatedTransaction(ctx, "init-migrate", func(conn *pop.Connection) error {
					// We do not want to remove the legacy migration version or subsequent migrations might be applied twice.
					//
					// Do not activate the following - it is just for reference.
					//
					// if _, err := tx.Store.Exec(fmt.Sprintf("DELETE FROM %s WHERE version = ?", mtn), legacyVersion); err != nil {
					//	return errors.Wrapf(err, "problem removing legacy version %s", mi.Version)
					// }

					// #nosec G201 - mtn is a system-wide const
					err := conn.RawQuery(fmt.Sprintf("INSERT INTO %s (version) VALUES (?)", mtn), mi.Version).Exec()
					return errors.Wrapf(err, "problem inserting migration version %s", mi.Version)
				}); err != nil {
					return errors.WithStack(err)
				}
				continue
			}

			l.Info("Migration has not yet been applied, running migration.")

			if err := mi.Valid(); err != nil {
				return errors.WithStack(err)
			}

			noTx := mb.shouldNotUseTransaction(mi)
			if noTx {
				l.Info("NOT running migrations inside a transaction")
				if err := mi.Runner(mi, c); err != nil {
					return errors.WithStack(err)
				}

				// #nosec G201 - mtn is a system-wide const
				if err := c.RawQuery(fmt.Sprintf("INSERT INTO %s (version) VALUES (?)", mtn), mi.Version).Exec(); err != nil {
					return errors.Wrapf(err, "problem inserting migration version %s. YOUR DATABASE MAY BE IN AN INCONSISTENT STATE! MANUAL INTERVENTION REQUIRED!", mi.Version)
				}
			} else {
				if err := mb.isolatedTransaction(ctx, "up", func(conn *pop.Connection) error {
					if err := mi.Runner(mi, conn); err != nil {
						return errors.WithStack(err)
					}

					// #nosec G201 - mtn is a system-wide const
					if err := conn.RawQuery(fmt.Sprintf("INSERT INTO %s (version) VALUES (?)", mtn), mi.Version).Exec(); err != nil {
						return errors.Wrapf(err, "problem inserting migration version %s", mi.Version)
					}
					return nil
				}); err != nil {
					return errors.WithStack(err)
				}
			}

			l.WithField("autocommit", noTx).Infof("> %s applied successfully", mi.Name)
			applied++
			if step > 0 && applied >= step {
				break
			}
		}
		if applied == 0 {
			mb.l.Infof("Migrations already up to date, nothing to apply")
		} else {
			mb.l.Infof("Successfully applied %d migrations.", applied)
		}

		// Save a SQLite template after applying all migrations (not for
		// partial step runs) so future test runs can skip migrations.
		//
		// Only save when at least one migration was applied. Saving when
		// applied == 0 either captures an empty DB or an already-migrated
		// one that may contain test data inserted by an earlier registry
		// (e.g. the Hydra janitor reuses the same DSN), which would
		// pollute the shared template.
		//
		// Use VACUUM INTO to produce a clean, fully-checkpointed copy
		// without a separate wal_checkpoint step. Write to a temp path
		// unique to this test's database, then atomically rename to the
		// shared template path. This prevents concurrent saves from
		// multiple parallel tests from corrupting the template:
		// os.Rename is atomic on the same filesystem, so the last writer
		// wins with valid content and no reader ever sees a partial file.
		if isOnDiskSQLite && step <= 0 && !mb.disableGoldenDatabase && applied > 0 {
			templatePath := mb.sqliteTemplatePath()
			tmp, err := os.CreateTemp(filepath.Dir(templatePath), ".sqlite-template-*.sqlite")
			if err != nil {
				mb.l.Errorf("Failed to allocate SQLite template file: err=%v", err)
				return nil
			}
			tmpPath := tmp.Name()
			if closeErr := tmp.Close(); closeErr != nil {
				_ = os.Remove(tmpPath)
				mb.l.Errorf("Failed to close SQLite template file: err=%v", closeErr)
				return nil
			}
			// VACUUM INTO requires that the destination does not exist.
			if removeErr := os.Remove(tmpPath); removeErr != nil {
				mb.l.Errorf("Failed to prepare SQLite template file: err=%v", removeErr)
				return nil
			}
			defer func() { _ = os.Remove(tmpPath) }()
			// #nosec G201 - tmpPath is under os.TempDir() with a
			// hex-hash filename, never derived from user input.
			vacuumSQL := fmt.Sprintf("VACUUM INTO '%s'", strings.ReplaceAll(tmpPath, "'", "''"))
			if err := mb.c.RawQuery(vacuumSQL).Exec(); err != nil {
				mb.l.Errorf("Failed to create SQLite template copy: err=%v", err)
			} else if err := os.Rename(tmpPath, templatePath); err != nil {
				_ = os.Remove(tmpPath)
				mb.l.Errorf("Failed to install SQLite template: src=%s dst=%s err=%v", tmpPath, templatePath, err)
			} else {
				mfs := mb.migrationsUp.sortAndFilter(mb.c.Dialect.Name(), mb.migrationFallbacks()...)
				mb.l.Infof("SQLite template saved, subsequent test runs will skip %d migrations: path=%s", len(mfs), templatePath)
			}
		}
		return nil
	})
	return applied, errors.WithStack(err)
}

// Down runs pending "down" migrations and rolls back the
// database by the specified number of steps.
// If step <= 0, all down migrations are run.
func (mb *MigrationBox) Down(ctx context.Context, steps int) (err error) {
	ctx, span := startSpan(ctx, MigrationDownOpName, trace.WithAttributes(attribute.Int("steps", steps)))
	defer otelx.End(span, &err)

	if steps <= 0 {
		steps = math.MaxInt
	}
	remainingSteps := steps

	c := mb.c.WithContext(ctx)
	return errors.WithStack(mb.exec(ctx, func() (err error) {
		mtn := sanitizedMigrationTableName(c)
		count, err := c.Count(mtn)
		if err != nil {
			return errors.Wrap(err, "migration down: unable count existing migration")
		}
		attemptSteps := min(remainingSteps, count)

		mfs := mb.migrationsDown.sortAndFilter(mb.c.Dialect.Name(), mb.migrationFallbacks()...)
		slices.Reverse(mfs)
		if len(mfs) > count {
			// skip all migrations that were not yet applied
			mfs = mfs[len(mfs)-count:]
		}

		reverted := 0
		defer func() {
			migrationsToRevertCount := min(attemptSteps, len(mfs))
			mb.l.Debugf("Successfully reverted %d/%d migrations.", reverted, migrationsToRevertCount)
			if err != nil {
				mb.l.WithError(err).Error("Problem reverting migrations.")
			}
		}()
		for i, mi := range mfs {
			if i >= attemptSteps {
				break
			}
			l := mb.l.WithField("version", mi.Version).WithField("migration_name", mi.Name).WithField("migration_file", mi.Path)
			l.Debugf("handling migration %s", mi.Name)
			exists, err := c.Where("version = ?", mi.Version).Exists(mtn)
			if err != nil {
				return errors.Wrapf(err, "problem checking for migration version %s", mi.Version)
			}

			if !exists && len(mi.Version) > 14 {
				legacyVersion := mi.Version[:14]
				legacyVersionExists, err := c.Where("version = ?", legacyVersion).Exists(mtn)
				if err != nil {
					return errors.Wrapf(err, "problem checking for legacy migration version %s", legacyVersion)
				}

				if !legacyVersionExists {
					return errors.Errorf("neither normal (%s) nor legacy migration (%s) exist", mi.Version, legacyVersion)
				}
			} else if !exists {
				return errors.Errorf("migration version %s does not exist", mi.Version)
			}

			if err := mi.Valid(); err != nil {
				return errors.WithStack(err)
			}

			if mb.shouldNotUseTransaction(mi) {
				err := mi.Runner(mi, c)
				if err != nil {
					return errors.WithStack(err)
				}

				// #nosec G201 - mtn is a system-wide const
				if err := c.RawQuery(fmt.Sprintf("DELETE FROM %s WHERE version = ?", mtn), mi.Version).Exec(); err != nil {
					return errors.Wrapf(err, "problem deleting migration version %s. YOUR DATABASE MAY BE IN AN INCONSISTENT STATE! MANUAL INTERVENTION REQUIRED!", mi.Version)
				}
			} else {
				if err := mb.isolatedTransaction(ctx, "down", func(conn *pop.Connection) error {
					err := mi.Runner(mi, conn)
					if err != nil {
						return errors.WithStack(err)
					}

					// #nosec G201 - mtn is a system-wide const
					if err := conn.RawQuery(fmt.Sprintf("DELETE FROM %s WHERE version = ?", mtn), mi.Version).Exec(); err != nil {
						return errors.Wrapf(err, "problem deleting migration version %s", mi.Version)
					}

					return nil
				}); err != nil {
					return errors.WithStack(err)
				}
			}

			l.Infof("%s applied successfully", mi.Name)
			reverted++
			remainingSteps--
		}
		return nil
	}))
}

func (mb *MigrationBox) createTransactionalMigrationTable(ctx context.Context, c *pop.Connection, l *logrusx.Logger) error {
	mtn := sanitizedMigrationTableName(c)

	if err := mb.createMigrationStatusTableTransaction(ctx, []string{
		fmt.Sprintf(`CREATE TABLE %s (version VARCHAR (48) NOT NULL, version_self INT NOT NULL DEFAULT 0)`, mtn),
		fmt.Sprintf(`CREATE UNIQUE INDEX %s_version_idx ON %s (version)`, mtn, mtn),
		fmt.Sprintf(`CREATE INDEX %s_version_self_idx ON %s (version_self)`, mtn, mtn),
	}); err != nil {
		return errors.WithStack(err)
	}

	l.WithField("migration_table", mtn).Debug("Transactional migration table created successfully.")

	return nil
}

func (mb *MigrationBox) migrateToTransactionalMigrationTable(ctx context.Context, c *pop.Connection, l *logrusx.Logger) error {
	// This means the new pop migrator has also not yet been applied, do that now.
	mtn := sanitizedMigrationTableName(c)

	withOn := fmt.Sprintf(" ON %s", mtn)
	if c.Dialect.Name() != "mysql" {
		withOn = ""
	}

	interimTable := fmt.Sprintf("%s_transactional", mtn)
	workload := [][]string{
		{
			fmt.Sprintf(`DROP INDEX %s_version_idx%s`, mtn, withOn),
			fmt.Sprintf(`CREATE TABLE %s (version VARCHAR (48) NOT NULL, version_self INT NOT NULL DEFAULT 0)`, interimTable),
			fmt.Sprintf(`CREATE UNIQUE INDEX %s_version_idx ON %s (version)`, mtn, interimTable),
			fmt.Sprintf(`CREATE INDEX %s_version_self_idx ON %s (version_self)`, mtn, interimTable),
			// #nosec G201 - mtn is a system-wide const
			fmt.Sprintf(`INSERT INTO %s (version) SELECT version FROM %s`, interimTable, mtn),
			fmt.Sprintf(`ALTER TABLE %s RENAME TO %s_pop_legacy`, mtn, mtn),
		},
		{
			fmt.Sprintf(`ALTER TABLE %s RENAME TO %s`, interimTable, mtn),
		},
	}

	if err := mb.createMigrationStatusTableTransaction(ctx, workload...); err != nil {
		return errors.WithStack(err)
	}

	l.WithField("migration_table", mtn).Debug("Successfully migrated legacy schema_migration to new transactional schema_migration table.")

	return nil
}

func (mb *MigrationBox) isolatedTransaction(ctx context.Context, direction string, fn func(c *pop.Connection) error) (err error) {
	ctx, span := startSpan(ctx, MigrationRunTransactionOpName, trace.WithAttributes(attribute.String("migration_direction", direction)))
	defer otelx.End(span, &err)

	if mb.perMigrationTimeout > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeoutCause(ctx, mb.perMigrationTimeout, errors.Errorf("failed to run all SQL migrations: direction=%s timeout=%s", direction, mb.perMigrationTimeout))
		defer cancel()
	}

	return Transaction(ctx, mb.c.WithContext(ctx), func(ctx context.Context, connection *pop.Connection) error {
		return errors.WithStack(fn(connection))
	})
}

func (mb *MigrationBox) createMigrationStatusTableTransaction(ctx context.Context, transactions ...[]string) error {
	for _, statements := range transactions {
		// CockroachDB does not support transactional schema changes, so we have to run
		// the statements outside of a transaction. The same applies to any dialect
		// that opts into autocommit DDL (see noTxDDL).
		if mb.noTxDDL() {
			for _, statement := range statements {
				if err := mb.c.WithContext(ctx).RawQuery(statement).Exec(); err != nil {
					return errors.Wrapf(err, "unable to execute statement: %s", statement)
				}
			}
		} else {
			if err := mb.isolatedTransaction(ctx, "init", func(conn *pop.Connection) error {
				for _, statement := range statements {
					if err := conn.WithContext(ctx).RawQuery(statement).Exec(); err != nil {
						return errors.Wrapf(err, "unable to execute statement: %s", statement)
					}
				}
				return nil
			}); err != nil {
				return errors.WithStack(err)
			}
		}
	}

	return nil
}

// CreateSchemaMigrations sets up a table to track migrations. This is an idempotent
// operation.
func (mb *MigrationBox) CreateSchemaMigrations(ctx context.Context) error {
	ctx, span := startSpan(ctx, MigrationInitOpName)
	defer span.End()

	c := mb.c.WithContext(ctx)

	mtn := sanitizedMigrationTableName(c)
	mb.l.WithField("migration_table", mtn).Debug("Checking if legacy migration table exists.")
	_, err := c.Store.Exec(fmt.Sprintf("select version from %s", mtn))
	if err != nil {
		mb.l.WithError(err).WithField("migration_table", mtn).Debug("An error occurred while checking for the legacy migration table, maybe it does not exist yet? Trying to create.")
		// This means that the legacy pop migrator has not yet been applied
		return errors.WithStack(mb.createTransactionalMigrationTable(ctx, c, mb.l))
	}

	mb.l.WithField("migration_table", mtn).Debug("A migration table exists, checking if it is a transactional migration table.")
	_, err = c.Store.Exec(fmt.Sprintf("select version, version_self from %s", mtn))
	if err != nil {
		mb.l.WithError(err).WithField("migration_table", mtn).Debug("An error occurred while checking for the transactional migration table, maybe it does not exist yet? Trying to create.")
		return errors.WithStack(mb.migrateToTransactionalMigrationTable(ctx, c, mb.l))
	}

	mb.l.WithField("migration_table", mtn).Debug("Migration tables exist and are up to date.")
	return nil
}

type MigrationStatus struct {
	State       string `json:"state"`
	Version     string `json:"version"`
	Name        string `json:"name"`
	ContentUp   string `json:"content"`
	ContentDown string `json:"content_down"`
}

type MigrationStatuses []MigrationStatus

var _ cmdx.Table = (MigrationStatuses)(nil)

func (m MigrationStatuses) Header() []string {
	return []string{"Version", "Name", "Status"}
}

func (m MigrationStatuses) Table() [][]string {
	t := make([][]string, len(m))
	for i, s := range m {
		t[i] = []string{s.Version, s.Name, s.State}
	}
	return t
}

func (m MigrationStatuses) Interface() interface{} {
	return m
}

func (m MigrationStatuses) Len() int {
	return len(m)
}

func (m MigrationStatuses) IDs() []string {
	ids := make([]string, len(m))
	for i, s := range m {
		ids[i] = s.Version
	}
	return ids
}

func (m MigrationStatuses) HasPending() bool {
	for _, mm := range m {
		if mm.State == Pending {
			return true
		}
	}
	return false
}

func sanitizedMigrationTableName(con *pop.Connection) string {
	return regexp.MustCompile(`\W`).ReplaceAllString(con.MigrationTableName(), "")
}

func errIsTableNotFound(err error) bool {
	return strings.Contains(err.Error(), "no such table:") || // sqlite
		strings.Contains(err.Error(), "Error 1146") || // MySQL
		strings.Contains(err.Error(), "SQLSTATE 42P01") // PostgreSQL / CockroachDB
}

// Status prints out the status of applied/pending migrations.
func (mb *MigrationBox) Status(ctx context.Context) (MigrationStatuses, error) {
	ctx, span := startSpan(ctx, MigrationStatusOpName)
	defer span.End()

	con := mb.c.WithContext(ctx)

	dialect := mb.c.Dialect.Name()
	fallbacks := mb.migrationFallbacks()
	migrationsUp := mb.migrationsUp.sortAndFilter(dialect, fallbacks...)

	if len(migrationsUp) == 0 {
		return nil, errors.Errorf("unable to find any migrations for dialect: %s", dialect)
	}

	alreadyApplied := make([]string, 0, len(migrationsUp))
	err := con.RawQuery(fmt.Sprintf("SELECT version FROM %s", sanitizedMigrationTableName(con))).All(&alreadyApplied)
	if err != nil {
		if errIsTableNotFound(err) {
			// This means that no migrations have been applied and we need to apply all of them first!
			//
			// It also means that we can ignore this state and act as if no migrations have been applied yet.
		} else {
			// On any other error, we fail.
			return nil, errors.Wrap(err, "problem with migration")
		}
	}

	statuses := make(MigrationStatuses, len(migrationsUp))
	for k, mf := range migrationsUp {
		downContent := "-- error: no down migration defined for this migration"
		if mDown := mb.migrationsDown.find(mf.Version, dialect, fallbacks...); mDown != nil {
			downContent = mDown.Content
		}
		statuses[k] = MigrationStatus{
			State:       Pending,
			Version:     mf.Version,
			Name:        mf.Name,
			ContentUp:   mf.Content,
			ContentDown: downContent,
		}

		if slices.ContainsFunc(alreadyApplied, func(applied string) bool {
			return applied == mf.Version || (len(mf.Version) > 14 && applied == mf.Version[:14])
		}) {
			statuses[k].State = Applied
			continue
		}
	}

	return statuses, nil
}

// DumpMigrationSchema will generate a file of the current database schema
func (mb *MigrationBox) DumpMigrationSchema(ctx context.Context) error {
	c := mb.c.WithContext(ctx)
	schema := "schema.sql"
	f, err := os.Create(schema) //#nosec:G304
	if err != nil {
		return errors.WithStack(err)
	}
	err = c.Dialect.DumpSchema(f)
	if err != nil {
		_ = os.RemoveAll(schema)
		return errors.WithStack(err)
	}
	return nil
}

func (mb *MigrationBox) exec(ctx context.Context, fn func() error) error {
	now := time.Now()
	defer mb.printTimer(now)

	err := mb.CreateSchemaMigrations(ctx)
	if err != nil {
		return errors.Wrap(err, "migrator: problem creating schema migrations")
	}

	if mb.c.Dialect.Name() == "sqlite3" {
		if err := mb.c.RawQuery("PRAGMA foreign_keys=OFF").Exec(); err != nil {
			return sqlcon.HandleError(err)
		}
	}

	switch mb.c.Dialect.Name() {
	case dbal.DriverCockroachDB, dbal.DriverYugabyteDB:
		outer := fn
		fn = func() error {
			// CreateSchemaMigrations runs before this wrapper. YugabyteDB uses
			// autocommit DDL, but its pgwire errors expose the same retryable
			// SQLSTATEs that crdb.Execute classifies for whole-run retries.
			return errors.WithStack(crdb.Execute(outer))
		}
	}

	if err := fn(); err != nil {
		return errors.WithStack(err)
	}

	if mb.c.Dialect.Name() == "sqlite3" {
		if err := mb.c.RawQuery("PRAGMA foreign_keys=ON").Exec(); err != nil {
			return sqlcon.HandleError(err)
		}
	}

	return nil
}

func (mb *MigrationBox) printTimer(timerStart time.Time) {
	diff := time.Since(timerStart).Seconds()
	if diff > 60 {
		mb.l.Debugf("%.4f minutes", diff/60)
	} else {
		mb.l.Debugf("%.4f seconds", diff)
	}
}