80 lines
2.5 KiB
Go
80 lines
2.5 KiB
Go
package data
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"gorm.io/gorm"
|
|
"kra/internal/config"
|
|
"kra/pkg/database/migration"
|
|
"kra/pkg/module"
|
|
)
|
|
|
|
// MigrateDatabase opens only the configured primary database. It does not
|
|
// create a database, start a scheduler, open SDK clients, or initialize users.
|
|
func MigrateDatabase(ctx context.Context, c *config.Database, catalog module.Catalog, logger *slog.Logger, checkOnly bool) error {
|
|
if !databaseConnectionConfigured(c) {
|
|
return fmt.Errorf("migration requires an existing configured database")
|
|
}
|
|
migrationConfig := *c
|
|
// One connection holds the advisory lock, a second executes migrations.
|
|
if !checkOnly && migrationConfig.MaxOpenConns > 0 && migrationConfig.MaxOpenConns < 2 {
|
|
migrationConfig.MaxOpenConns = 2
|
|
}
|
|
db, err := openDatabase(&migrationConfig, false, "", logger)
|
|
if err != nil {
|
|
return fmt.Errorf("open migration database: %w", err)
|
|
}
|
|
defer closeGormDB(db)
|
|
db = db.WithContext(ctx)
|
|
if checkOnly {
|
|
return checkMigratedSchema(db, catalog)
|
|
}
|
|
// A dedicated connection owns the lock even when a migration uses a
|
|
// transaction or temporarily changes connection-local SQL compatibility.
|
|
run := func() error {
|
|
if err := catalog.CheckPrerequisites(db); err != nil {
|
|
return err
|
|
}
|
|
if err := migrateAll(db, catalog); err != nil {
|
|
return err
|
|
}
|
|
return checkMigratedSchema(db, catalog)
|
|
}
|
|
if db.Dialector.Name() != "mysql" {
|
|
return run()
|
|
}
|
|
return db.Connection(func(conn *gorm.DB) error {
|
|
var name string
|
|
if err := conn.Raw("SELECT CONCAT('sw_to_kra:', MD5(DATABASE()))").Scan(&name).Error; err != nil {
|
|
return err
|
|
}
|
|
var locked int
|
|
if err := conn.Session(&gorm.Session{NewDB: true}).Raw("SELECT GET_LOCK(?, 0)", name).Scan(&locked).Error; err != nil {
|
|
return err
|
|
}
|
|
if locked != 1 {
|
|
return fmt.Errorf("another migration owns the database lock")
|
|
}
|
|
defer func() {
|
|
cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
|
defer cancel()
|
|
conn.Session(&gorm.Session{NewDB: true}).WithContext(cleanup).Exec("DO RELEASE_LOCK(?)", name)
|
|
}()
|
|
return run()
|
|
})
|
|
}
|
|
|
|
func checkMigratedSchema(db *gorm.DB, catalog module.Catalog) error {
|
|
db = db.Session(&gorm.Session{NewDB: true})
|
|
if err := migration.CheckApplied(db, catalog.MigrationSteps()); err != nil {
|
|
return fmt.Errorf("run -migrate before starting the service: %w", err)
|
|
}
|
|
if err := catalog.CheckSchema(db); err != nil {
|
|
return fmt.Errorf("database schema is incomplete: %w", err)
|
|
}
|
|
return nil
|
|
}
|