427 lines
13 KiB
Go
427 lines
13 KiB
Go
package swmigration
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"gorm.io/gorm"
|
|
scripts "kra/deploy/migrations"
|
|
"kra/pkg/database/migration"
|
|
"kra/pkg/module"
|
|
)
|
|
|
|
// Recognized identifies an old SW database or a previous automatic upgrade.
|
|
// Database names are not used to infer ownership of unrelated schemas.
|
|
func Recognized(db *gorm.DB) (bool, error) {
|
|
if db.Dialector.Name() != "mysql" {
|
|
return false, nil
|
|
}
|
|
tables, err := tableSet(db)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if tables["sw_kra_state"] || tables["sys_user"] && tables["tab_sw_report_day"] && tables["tab_sw_report_server_day"] {
|
|
return true, nil
|
|
}
|
|
if !tables[TableName] {
|
|
return false, nil
|
|
}
|
|
// The state table also exists in new template installations. Only a
|
|
// recorded legacy workflow identifies a previously upgraded SW database.
|
|
var count int64
|
|
if err := db.Table(TableName).Where("name=?", workflowName).Count(&count).Error; err != nil {
|
|
return false, err
|
|
}
|
|
return count > 0, nil
|
|
}
|
|
|
|
func tableSet(db *gorm.DB) (map[string]bool, error) {
|
|
var names []string
|
|
if err := db.Raw("SELECT table_name FROM information_schema.tables WHERE table_schema=DATABASE() AND table_type='BASE TABLE'").Scan(&names).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
set := map[string]bool{}
|
|
for _, name := range names {
|
|
set[name] = true
|
|
}
|
|
return set, nil
|
|
}
|
|
|
|
func Run(ctx context.Context, db *gorm.DB, catalog module.Catalog, backupRoot string, logger *slog.Logger) (runErr error) {
|
|
if logger == nil {
|
|
logger = slog.Default()
|
|
}
|
|
db = db.Session(&gorm.Session{NewDB: true}).WithContext(ctx)
|
|
pool, err := db.DB()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
conn, err := pool.Conn(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer conn.Close()
|
|
var schema, serverUUID string
|
|
if err = conn.QueryRowContext(ctx, "SELECT DATABASE(),@@server_uuid").Scan(&schema, &serverUUID); err != nil {
|
|
return err
|
|
}
|
|
if schema == "" {
|
|
return fmt.Errorf("automatic upgrade requires a selected database")
|
|
}
|
|
var lockName string
|
|
if err = conn.QueryRowContext(ctx, "SELECT CONCAT('sw_to_kra:',MD5(DATABASE()))").Scan(&lockName); err != nil {
|
|
return err
|
|
}
|
|
for {
|
|
var locked sql.NullInt64
|
|
if err = conn.QueryRowContext(ctx, "SELECT GET_LOCK(?,10)", lockName).Scan(&locked); err != nil {
|
|
return err
|
|
}
|
|
if locked.Valid && locked.Int64 == 1 {
|
|
break
|
|
}
|
|
if !locked.Valid {
|
|
return fmt.Errorf("database migration lock is unavailable")
|
|
}
|
|
logger.Info("waiting for another database upgrade", "database", schema)
|
|
}
|
|
defer func() {
|
|
clean, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
|
defer cancel()
|
|
_, _ = conn.ExecContext(clean, "DO RELEASE_LOCK(?)", lockName)
|
|
}()
|
|
if err = StateMigration().Migrate(db); err != nil {
|
|
return err
|
|
}
|
|
state := statePO{Name: workflowName}
|
|
err = db.First(&state, "name=?", workflowName).Error
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
state = statePO{Name: workflowName, Version: workflowVersion, Phase: "backup", DatabaseIdentity: serverUUID + ":" + schema, StartedAt: time.Now(), UpdatedAt: time.Now()}
|
|
initial, readErr := tableSet(db)
|
|
if readErr != nil {
|
|
return readErr
|
|
}
|
|
state.SourceWasLegacy = initial["sys_user"]
|
|
if initial["sw_kra_state"] {
|
|
var phase string
|
|
if err := db.Table("sw_kra_state").Select("phase").Where("id=1").Scan(&phase).Error; err != nil {
|
|
return err
|
|
}
|
|
state.SourceWasLegacy = state.SourceWasLegacy || phase != "completed"
|
|
}
|
|
if err = db.Create(&state).Error; err != nil {
|
|
return err
|
|
}
|
|
} else if err != nil {
|
|
return err
|
|
}
|
|
if state.Version != workflowVersion || state.DatabaseIdentity != serverUUID+":"+schema {
|
|
return fmt.Errorf("automatic upgrade identity/version does not match this database")
|
|
}
|
|
started := time.Now()
|
|
logger.Info("automatic business database upgrade started", "database", schema, "phase", state.Phase)
|
|
defer func() {
|
|
status := "completed"
|
|
if runErr != nil {
|
|
status = "failed"
|
|
clean, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
|
defer cancel()
|
|
_ = db.Session(&gorm.Session{NewDB: true}).WithContext(clean).Model(&statePO{}).Where("name=?", workflowName).Updates(map[string]any{"last_error": runErr.Error(), "updated_at": time.Now()}).Error
|
|
}
|
|
logger.Info("automatic business database upgrade finished", "database", schema, "status", status, "elapsed_ms", time.Since(started).Milliseconds())
|
|
}()
|
|
save := func() error { state.UpdatedAt = time.Now(); return db.Save(&state).Error }
|
|
stage := func(name string, work func() error) error {
|
|
state.Phase = name
|
|
state.LastError = ""
|
|
if err := save(); err != nil {
|
|
return err
|
|
}
|
|
begin := time.Now()
|
|
logger.Info("database upgrade stage started", "stage", name)
|
|
done := make(chan struct{})
|
|
stopped := make(chan struct{})
|
|
go func() {
|
|
defer close(stopped)
|
|
ticker := time.NewTicker(10 * time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-done:
|
|
return
|
|
case <-ticker.C:
|
|
logger.Info("database upgrade is still running", "stage", name, "elapsed_ms", time.Since(begin).Milliseconds())
|
|
}
|
|
}
|
|
}()
|
|
err := work()
|
|
close(done)
|
|
<-stopped
|
|
if err != nil {
|
|
return fmt.Errorf("database upgrade %s: %w", name, err)
|
|
}
|
|
logger.Info("database upgrade stage completed", "stage", name, "elapsed_ms", time.Since(begin).Milliseconds())
|
|
return nil
|
|
}
|
|
resumingCleanup := state.CleanupBackupPath != "" && state.VerifiedAt != nil
|
|
wasComplete := state.CompletedAt != nil
|
|
if !resumingCleanup && !wasComplete {
|
|
if state.BackupPath == "" {
|
|
if err := stage("backup", func() error {
|
|
var e error
|
|
state.BackupPath, state.BackupSHA256, e = backupDatabase(ctx, db, backupRoot, state.DatabaseIdentity, "source", nil, logger)
|
|
return e
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
if err := save(); err != nil {
|
|
return err
|
|
}
|
|
} else if err := verifyBackup(state.BackupPath, state.BackupSHA256); err != nil {
|
|
return err
|
|
}
|
|
if err := stage("framework", func() error { return migrateFramework(ctx, db, conn, logger) }); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err := stage("migrations", func() error {
|
|
if err := catalog.CheckPrerequisites(db); err != nil {
|
|
return err
|
|
}
|
|
return migration.Run(db, catalog.MigrationSteps())
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
if err := stage("verify", func() error {
|
|
if err := migration.CheckApplied(db, catalog.MigrationSteps()); err != nil {
|
|
return err
|
|
}
|
|
return catalog.CheckSchema(db)
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
remaining, err := remainingTargets(db)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(remaining) > 0 {
|
|
if wasComplete {
|
|
return fmt.Errorf("retired tables reappeared after a completed upgrade; automatic deletion refused")
|
|
}
|
|
if !resumingCleanup && !wasComplete {
|
|
if err := verifyLegacyChecks(db); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if state.SourceWasLegacy && !state.TasksRestored {
|
|
if err := stage("tasks", func() error {
|
|
return db.Transaction(func(tx *gorm.DB) error {
|
|
if err := tx.Exec("UPDATE sys_timed_tasks t JOIN sw_kra_id_map m ON m.entity='task' AND m.new_id=t.id JOIN sj_job j ON CAST(j.id AS CHAR)=m.old_id SET t.enabled=(j.job_status=1) WHERE j.deleted=0").Error; err != nil {
|
|
return err
|
|
}
|
|
return tx.Model(&statePO{}).Where("name=?", workflowName).Update("tasks_restored", true).Error
|
|
})
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
state.TasksRestored = true
|
|
}
|
|
if state.CleanupBackupPath == "" {
|
|
if err := stage("cleanup-backup", func() error {
|
|
var e error
|
|
state.CleanupBackupPath, state.CleanupSHA256, e = backupDatabase(ctx, db, backupRoot, state.DatabaseIdentity, "cleanup", remaining, logger)
|
|
return e
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
if err := save(); err != nil {
|
|
return err
|
|
}
|
|
} else if err := verifyBackup(state.CleanupBackupPath, state.CleanupSHA256); err != nil {
|
|
return err
|
|
}
|
|
// The journal survives deletion of all old migration helpers. A restart
|
|
// can finish a partially completed DROP list using the verified archive.
|
|
now := time.Now()
|
|
state.VerifiedAt = &now
|
|
state.VerifiedMigrations = migrationHash(catalog.MigrationSteps())
|
|
if err := stage("cleanup", func() error {
|
|
// CALL can commit before a killed process reaches DROP PROCEDURE.
|
|
// Remove only our own shipped routine before checking dependencies.
|
|
if _, err := conn.ExecContext(ctx, "DROP PROCEDURE IF EXISTS sw_kra_migrate_20261103"); err != nil {
|
|
return err
|
|
}
|
|
return executeScript(ctx, conn, scripts.CleanupSQL(), true, logger)
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err := migration.CheckApplied(db, catalog.MigrationSteps()); err != nil {
|
|
return err
|
|
}
|
|
if err := catalog.CheckSchema(db); err != nil {
|
|
return err
|
|
}
|
|
remaining, err = remainingTargets(db)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(remaining) != 0 {
|
|
return fmt.Errorf("legacy cleanup left %d tables", len(remaining))
|
|
}
|
|
now := time.Now()
|
|
state.Phase = "complete"
|
|
if state.CompletedAt == nil {
|
|
state.CompletedAt = &now
|
|
}
|
|
state.VerifiedAt = &now
|
|
state.LastError = ""
|
|
state.VerifiedMigrations = migrationHash(catalog.MigrationSteps())
|
|
return save()
|
|
}
|
|
|
|
func migrationHash(steps []migration.Step) string {
|
|
ids := make([]string, len(steps))
|
|
for i, step := range steps {
|
|
ids[i] = step.ID
|
|
}
|
|
sort.Strings(ids)
|
|
sum := sha256.Sum256([]byte(strings.Join(ids, "\n")))
|
|
return hex.EncodeToString(sum[:])
|
|
}
|
|
|
|
func remainingTargets(db *gorm.DB) (map[string]bool, error) {
|
|
tables, err := tableSet(db)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
remaining := map[string]bool{}
|
|
for _, name := range scripts.CleanupTables() {
|
|
if tables[name] {
|
|
remaining[name] = true
|
|
}
|
|
}
|
|
return remaining, nil
|
|
}
|
|
|
|
func executeScript(ctx context.Context, conn *sql.Conn, source string, cleanup bool, logger *slog.Logger) error {
|
|
statements, err := scripts.Statements(source)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, statement := range statements {
|
|
upper := strings.ToUpper(strings.TrimSpace(statement))
|
|
if strings.HasPrefix(upper, "USE ") {
|
|
continue
|
|
}
|
|
if cleanup && strings.HasPrefix(upper, "SET @SW_CLEANUP_EXECUTE") {
|
|
statement = "SET @sw_cleanup_execute=1"
|
|
}
|
|
rows, err := conn.QueryContext(ctx, statement)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for {
|
|
columns, err := rows.Columns()
|
|
if err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
for rows.Next() {
|
|
values := make([]any, len(columns))
|
|
dest := make([]any, len(columns))
|
|
for i := range values {
|
|
dest[i] = &values[i]
|
|
}
|
|
if err := rows.Scan(dest...); err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
// Result sets contain progress/metadata only, never copied rows.
|
|
if len(columns) > 0 && (columns[0] == "dropped_table" || columns[0] == "result") {
|
|
attrs := make([]any, 0, len(columns)*2)
|
|
for i, name := range columns {
|
|
value := values[i]
|
|
if b, ok := value.([]byte); ok {
|
|
value = string(b)
|
|
}
|
|
attrs = append(attrs, name, value)
|
|
}
|
|
logger.Info("database migration SQL result", attrs...)
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
if !rows.NextResultSet() {
|
|
break
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
rows.Close()
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func verifyLegacyChecks(db *gorm.DB) error {
|
|
tables, err := tableSet(db)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !tables["sw_kra_state"] || !tables["sw_kra_checks"] {
|
|
return fmt.Errorf("legacy migration evidence is missing; cleanup refused")
|
|
}
|
|
var completed int64
|
|
if err := db.Table("sw_kra_state").Where("id=1 AND version=? AND phase='completed' AND completed_at IS NOT NULL", "20261103_framework_migration_v5").Count(&completed).Error; err != nil {
|
|
return err
|
|
}
|
|
if completed != 1 {
|
|
return fmt.Errorf("legacy framework migration is incomplete")
|
|
}
|
|
if err := verifyCheckRows(db); err != nil {
|
|
return err
|
|
}
|
|
if tables["legacy_sw_20261103_sys_user"] && tables["sw_kra_id_map"] {
|
|
var archived, mapped int64
|
|
if err := db.Table("legacy_sw_20261103_sys_user").Where("del_flag='0'").Count(&archived).Error; err != nil {
|
|
return err
|
|
}
|
|
if err := db.Table("sw_kra_id_map").Where("entity='user'").Count(&mapped).Error; err != nil {
|
|
return err
|
|
}
|
|
if archived != mapped {
|
|
return fmt.Errorf("legacy accounts changed during conversion; cleanup refused")
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func verifyCheckRows(db *gorm.DB) error {
|
|
var invalid, backups, checks int64
|
|
if err := db.Table("sw_kra_checks").Where("passed<>1 OR expected_value IS NULL OR actual_value IS NULL OR expected_value<>actual_value").Count(&invalid).Error; err != nil {
|
|
return err
|
|
}
|
|
if err := db.Table("sw_kra_checks").Where("stage='backup'").Distinct("check_name").Count(&backups).Error; err != nil {
|
|
return err
|
|
}
|
|
if err := db.Table("sw_kra_checks").Where("stage='migration'").Distinct("check_name").Count(&checks).Error; err != nil {
|
|
return err
|
|
}
|
|
if invalid != 0 || backups != 52 || checks != 24 {
|
|
return fmt.Errorf("legacy migration validation is incomplete; cleanup refused")
|
|
}
|
|
return nil
|
|
}
|