package swmigration_test import ( "context" "database/sql" "encoding/json" "fmt" "io" "log/slog" "os" "path/filepath" "reflect" "sort" "strings" "sync" "testing" "time" driver "github.com/go-sql-driver/mysql" "golang.org/x/crypto/bcrypt" "gorm.io/driver/mysql" "gorm.io/gorm" "gorm.io/gorm/logger" assets "kra/deploy/migrations" biz "kra/internal/biz/business" bizsystem "kra/internal/biz/system" "kra/internal/config" rootdata "kra/internal/data" data "kra/internal/data/business" "kra/internal/data/swmigration" "kra/internal/integration/storage" "kra/internal/modules" "kra/pkg/database/migration" ) func localDatabase(t *testing.T) *gorm.DB { t.Helper() dsn := os.Getenv("SW_TEST_MYSQL_DSN") if dsn == "" { t.Skip("set SW_TEST_MYSQL_DSN for isolated MySQL upgrade tests") } cfg, err := driver.ParseDSN(dsn) if err != nil { t.Fatal(err) } if cfg.Net != "tcp" || !strings.HasPrefix(cfg.Addr, "127.0.0.1:") { t.Fatal("upgrade tests require an isolated local MySQL") } cfg.DBName = "" cfg.ParseTime = true cfg.Loc = time.FixedZone("Asia/Shanghai", 8*3600) admin, err := sql.Open("mysql", cfg.FormatDSN()) if err != nil { t.Fatal(err) } name := fmt.Sprintf("sw_test_upgrade_%d", time.Now().UnixNano()) if _, err := admin.Exec("CREATE DATABASE `" + name + "` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"); err != nil { admin.Close() t.Fatal(err) } cfg.DBName = name db, err := gorm.Open(mysql.Open(cfg.FormatDSN()), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)}) if err != nil { t.Fatal(err) } pool, _ := db.DB() pool.SetMaxOpenConns(4) t.Cleanup(func() { pool.Close() _, err := admin.Exec("DROP DATABASE `" + name + "`") if err != nil { t.Error(err) } admin.Close() }) return db } func legacyFixture(t *testing.T, db *gorm.DB) string { t.Helper() raw, err := os.ReadFile("testdata/legacy-schema.sql") if err != nil { t.Fatal(err) } statements, err := assets.Statements(string(raw)) if err != nil { t.Fatal(err) } for _, statement := range statements { if err := db.Exec(statement).Error; err != nil { t.Fatal(err) } } var surface struct{ Menus, Jobs []map[string]any } file, err := os.Open("testdata/legacy-surface.json") if err != nil { t.Fatal(err) } decoder := json.NewDecoder(file) decoder.UseNumber() err = decoder.Decode(&surface) file.Close() if err != nil { t.Fatal(err) } password, err := bcrypt.GenerateFromPassword([]byte("Fixture-password-2026"), bcrypt.MinCost) if err != nil { t.Fatal(err) } roleIDs := []string{"1", "2083497078545207297", "2087435436573081601", "2087361690139152385"} userIDs := []string{"9007199254740993", "9007199254740995", "9007199254740997", "9007199254740999"} for level, id := range roleIDs { key := fmt.Sprintf("business%d", level) if level == 0 { key = "superadmin" } insertLegacy(t, db, "sys_role", map[string]any{"role_id": id, "role_name": fmt.Sprintf("renamed-role-%d", level), "role_key": key, "role_sort": level, "data_scope": "1", "status": "0", "del_flag": "0", "create_time": "2026-01-01 08:00:00", "update_time": "2026-01-01 08:00:00"}) parent := "0" path := "0" if level > 1 { parent = userIDs[level-1] path = "0," + strings.Join(userIDs[1:level], ",") } insertLegacy(t, db, "sys_user", map[string]any{"user_id": userIDs[level], "user_name": fmt.Sprintf("fixture-user-%d", level), "nick_name": fmt.Sprintf("fixture-%d", level), "password": string(password), "status": "0", "del_flag": "0", "level": level, "parent_id": parent, "id_path": path, "create_time": "2026-01-01 08:00:00", "update_time": "2026-01-01 08:00:00"}) insertLegacy(t, db, "sys_user_role", map[string]any{"user_id": userIDs[level], "role_id": id}) insertLegacy(t, db, "sys_user_preference", map[string]any{"user_id": userIDs[level], "include_sub_promote": 1, "idcard_dedup": 1, "update_time": "2026-01-01 08:00:00"}) } for _, menu := range surface.Menus { insertLegacy(t, db, "sys_menu", menu) for _, id := range roleIDs { insertLegacy(t, db, "sys_role_menu", map[string]any{"role_id": id, "menu_id": menu["menu_id"]}) } } for i, job := range surface.Jobs { job["id"] = i + 1 job["job_name"] = fmt.Sprintf("fixture-job-%d", i) job["biz_id"] = fmt.Sprintf("fixture-job-%d", i) job["namespace_id"] = "fixture" job["args_str"] = "{}" job["deleted"] = 0 insertLegacy(t, db, "sj_job", job) } insertLegacy(t, db, "sys_user_bind_platform", map[string]any{"user_id": userIDs[3], "source_db": "lx", "platform_busier_id": 14, "platform_busier_account": "fixture-sdk", "status": "0", "create_time": "2026-01-01 08:00:00"}) insertLegacy(t, db, "sys_oss", map[string]any{"oss_id": "9007199254741093", "file_name": "fixture.png", "original_name": "fixture.png", "file_suffix": "png", "url": "https://example.invalid/fixture.png", "ext1": "{}", "create_by": userIDs[3], "create_time": "2026-01-01 08:00:00"}) insertLegacy(t, db, "sys_alert_rule", map[string]any{"id": 1, "rule_name": "fixture-rule", "rule_type": "fixture", "monitor_metric": 1, "monitor_scope": 1, "warning_color": "red", "creator_id": userIDs[3], "enabled": 1, "is_deleted": 0, "created_at": "2026-01-01 08:00:00", "updated_at": "2026-01-01 08:00:00"}) insertLegacy(t, db, "sys_alert_record", map[string]any{"id": "9007199254740993", "platform_code": "lx", "platform_name": "fixture", "rule_id": 1, "rule_name": "fixture-rule", "rule_type": "fixture", "target_promote_id": 100, "business_id": userIDs[3], "business_name": "fixture", "status": 1, "alert_date": "2026-01-01", "created_at": "2026-01-01 08:00:00", "updated_at": "2026-01-01 08:00:00"}) insertLegacy(t, db, "sys_alert_operation_log", map[string]any{"id": 1, "alert_id": "9007199254740993", "operator_id": userIDs[3], "log_type": 1, "attachment_ids": "[9007199254741093]", "created_at": "2026-01-01 08:00:00"}) insertLegacy(t, db, "tab_promote", map[string]any{"id": 100, "source_db": "lx", "account": "fixture-channel", "promote_level": 1, "busier_id": 14}) insertLegacy(t, db, "tab_user", map[string]any{"id": 1, "source_db": "lx", "account": "fixture-player", "promote_id": 100, "puid": 0, "register_time": "1767225600"}) insertLegacy(t, db, "tab_spend", map[string]any{"id": 1, "source_db": "lx", "user_id": 1, "promote_id": 100, "pay_status": 1, "pay_amount": "12.34", "pay_time": 1767225600, "pay_order_number": "fixture-order"}) if err := db.Exec("ALTER TABLE tab_promote AUTO_INCREMENT=2147483640").Error; err != nil { t.Fatal(err) } return string(password) } func insertLegacy(t *testing.T, db *gorm.DB, table string, values map[string]any) { t.Helper() var columns []struct { ColumnName, DataType, Extra, IsNullable string ColumnDefault *string } if err := db.Raw("SELECT column_name,data_type,extra,is_nullable,column_default FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name=? ORDER BY ordinal_position", table).Scan(&columns).Error; err != nil { t.Fatal(err) } known := map[string]bool{} for _, column := range columns { known[column.ColumnName] = true if _, exists := values[column.ColumnName]; exists || column.ColumnDefault != nil || column.IsNullable == "YES" || strings.Contains(column.Extra, "auto_increment") || strings.Contains(column.Extra, "GENERATED") { continue } var value any = "fixture" switch column.DataType { case "tinyint", "smallint", "mediumint", "int", "bigint", "decimal", "double", "float", "bit": value = 0 case "datetime", "timestamp": value = "2026-01-01 08:00:00" case "date": value = "2026-01-01" case "time": value = "00:00:00" case "json": value = "{}" } values[column.ColumnName] = value } var names []string for name := range values { if !known[name] { t.Fatalf("unknown fixture column %s.%s", table, name) } names = append(names, name) } sort.Strings(names) args := make([]any, len(names)) marks := make([]string, len(names)) quoted := make([]string, len(names)) for i, name := range names { args[i] = values[name] marks[i] = "?" quoted[i] = "`" + name + "`" } if err := db.Exec("INSERT INTO `"+table+"` ("+strings.Join(quoted, ",")+") VALUES ("+strings.Join(marks, ",")+")", args...).Error; err != nil { t.Fatalf("seed %s: %v", table, err) } } type provider struct{ db *gorm.DB } func (p provider) DB() *gorm.DB { return p.db } func assertComplete(t *testing.T, db *gorm.DB, password string) { t.Helper() if err := modules.Catalog().CheckSchema(db); err != nil { t.Fatal(err) } for _, name := range assets.CleanupTables() { if db.Migrator().HasTable(name) { t.Fatalf("retired table remains: %s", name) } } var phase string db.Table(swmigration.TableName).Select("phase").Where("name='legacy-sw'").Scan(&phase) if phase != "complete" { t.Fatalf("phase=%q", phase) } var users []struct { ID, AuthorityID uint Password string } if err := db.Table("sys_users").Order("id").Find(&users).Error; err != nil { t.Fatal(err) } if len(users) != 4 { t.Fatalf("migrated users=%d", len(users)) } for _, user := range users { if user.Password != password { t.Fatal("password bytes changed") } } var count int64 db.Raw("SELECT COUNT(*) FROM information_schema.tables WHERE table_schema=DATABASE() AND table_type='BASE TABLE'").Scan(&count) if count != 78 { t.Fatalf("final table count=%d, want 78", count) } db.Table("tab_user").Count(&count) if count != 1 { t.Fatal("raw user data changed") } var next uint64 db.Raw("SELECT auto_increment FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name='tab_promote'").Scan(&next) if next != 2147483640 { t.Fatalf("auto-increment changed: %d", next) } var surface struct{ Jobs []map[string]any } fixture, err := os.Open("testdata/legacy-surface.json") if err != nil { t.Fatal(err) } decoder := json.NewDecoder(fixture) decoder.UseNumber() err = decoder.Decode(&surface) fixture.Close() if err != nil { t.Fatal(err) } expectedEnabled := int64(2) // new incremental and full effectiveness jobs for _, job := range surface.Jobs { if fmt.Sprint(job["job_status"]) == "1" { expectedEnabled++ } } db.Table("sys_timed_tasks").Where("deleted_at IS NULL AND enabled=1").Count(&count) if count != expectedEnabled { t.Fatalf("legacy task enable states not restored: got %d want %d", count, expectedEnabled) } var attachments string db.Table("sys_alert_operation_log").Select("attachment_ids").Where("id=1").Scan(&attachments) if attachments == "[9007199254741093]" || attachments == "[]" { t.Fatalf("attachment migration did not finish before cleanup: %s", attachments) } var id uint uc := biz.NewAccountUsecase(data.NewAccountRepo(provider{db})) for level := 1; level <= 3; level++ { actor := bizsystem.Actor{UserID: users[0].ID, AuthorityID: 888} if level > 1 { var aid uint db.Table("sys_users").Select("authority_id").Where("id=?", id).Scan(&aid) actor = bizsystem.Actor{UserID: id, AuthorityID: aid} } created, err := uc.CreateChild(bizsystem.NewActorContext(context.Background(), actor), &biz.Account{Username: fmt.Sprintf("after-cleanup-%d", level), Password: "fixture-password", RealName: "fixture"}) if err != nil { t.Fatalf("create business level %d after map removal: %v", level, err) } id = created.ID if created.Level != level { t.Fatalf("created level %d instead of %d", created.Level, level) } } } func TestMySQLAutomaticUpgradeAndRestart(t *testing.T) { db := localDatabase(t) password := legacyFixture(t, db) root := t.TempDir() log := slog.New(slog.NewTextHandler(io.Discard, nil)) if err := swmigration.Run(context.Background(), db, modules.Catalog(), root, log); err != nil { t.Fatal(err) } assertComplete(t, db, password) archives, err := filepath.Glob(filepath.Join(root, "*.sql.gz")) if err != nil || len(archives) != 2 { t.Fatalf("backup archives=%v %v", archives, err) } if err := swmigration.Run(context.Background(), db, modules.Catalog(), root, log); err != nil { t.Fatalf("restart after cleanup: %v", err) } var count int64 db.Table("sys_users").Count(&count) if count != 7 { t.Fatal("restart reset created users") } } func TestMySQLUpgradeRecognitionRequiresLegacyEvidence(t *testing.T) { db := localDatabase(t) if err := swmigration.StateMigration().Migrate(db); err != nil { t.Fatal(err) } assertRecognized := func(want bool) { t.Helper() got, err := swmigration.Recognized(db) if err != nil || got != want { t.Fatalf("legacy recognition = %v, want %v: %v", got, want, err) } } // Every fresh template installation creates this table, even when it has // never contained legacy SW data or started the legacy upgrade workflow. assertRecognized(false) if err := db.Exec("INSERT INTO sys_business_upgrade_state(name,version,phase,database_identity) VALUES('another-workflow','v1','complete','fixture')").Error; err != nil { t.Fatal(err) } assertRecognized(false) if err := db.Exec("UPDATE sys_business_upgrade_state SET name='legacy-sw'").Error; err != nil { t.Fatal(err) } assertRecognized(true) } func TestMySQLUpgradeResumesAfterIndexFailure(t *testing.T) { db := localDatabase(t) password := legacyFixture(t, db) root := t.TempDir() log := slog.New(slog.NewTextHandler(io.Discard, nil)) if err := db.Exec("CREATE INDEX idx_sw_user_registration ON tab_user(id)").Error; err != nil { t.Fatal(err) } if err := swmigration.Run(context.Background(), db, modules.Catalog(), root, log); err == nil { t.Fatal("incompatible existing index accepted") } if !db.Migrator().HasTable("sw_kra_id_map") || !db.Migrator().HasTable("legacy_sw_20261103_sys_user") { t.Fatal("failed index migration deleted recovery data") } if err := db.Exec("DROP INDEX idx_sw_user_registration ON tab_user").Error; err != nil { t.Fatal(err) } if err := swmigration.Run(context.Background(), db, modules.Catalog(), root, log); err != nil { t.Fatalf("resume indexes: %v", err) } assertComplete(t, db, password) } func TestMySQLUpgradeResumesPartialCleanupOnlyWithVerifiedBackup(t *testing.T) { db := localDatabase(t) password := legacyFixture(t, db) root := t.TempDir() log := slog.New(slog.NewTextHandler(io.Discard, nil)) if err := db.Exec("CREATE VIEW retained_job_view AS SELECT id FROM sj_job").Error; err != nil { t.Fatal(err) } if err := swmigration.Run(context.Background(), db, modules.Catalog(), root, log); err == nil { t.Fatal("cleanup ignored a dependent view") } var path string if err := db.Table(swmigration.TableName).Select("cleanup_backup_path").Where("name='legacy-sw'").Scan(&path).Error; err != nil { t.Fatal(err) } original, err := os.ReadFile(path) if err != nil { t.Fatal(err) } // Simulate a crash after an earlier table was dropped. Its data is already // in the completed cleanup archive, so restarting must not replace it with // a new snapshot of only the tables that remain. if err := db.Exec("DROP TABLE bak_sw2kra_20261103_sys_user").Error; err != nil { t.Fatal(err) } if err := os.WriteFile(path, append(original, byte(1)), 0600); err != nil { t.Fatal(err) } if err := swmigration.Run(context.Background(), db, modules.Catalog(), root, log); err == nil || !strings.Contains(err.Error(), "checksum") { t.Fatalf("corrupt archive allowed deletion: %v", err) } if !db.Migrator().HasTable("sw_kra_id_map") { t.Fatal("mapping deleted without verified backup") } if err := os.WriteFile(path, original, 0600); err != nil { t.Fatal(err) } if err := db.Exec("DROP VIEW retained_job_view").Error; err != nil { t.Fatal(err) } if err := swmigration.Run(context.Background(), db, modules.Catalog(), root, log); err != nil { t.Fatalf("resume cleanup: %v", err) } assertComplete(t, db, password) } func TestMySQLConcurrentAutomaticUpgrades(t *testing.T) { db := localDatabase(t) password := legacyFixture(t, db) root := t.TempDir() log := slog.New(slog.NewTextHandler(io.Discard, nil)) errors := make(chan error, 2) var group sync.WaitGroup for i := 0; i < 2; i++ { group.Add(1) go func() { defer group.Done() errors <- swmigration.Run(context.Background(), db, modules.Catalog(), root, log) }() } group.Wait() close(errors) for err := range errors { if err != nil { t.Fatal(err) } } assertComplete(t, db, password) archives, _ := filepath.Glob(filepath.Join(root, "*.sql.gz")) if len(archives) != 2 { t.Fatalf("concurrent instances duplicated backups: %v", archives) } } type cancelOnProgress struct { slog.Handler cancel context.CancelFunc once *sync.Once } func (h cancelOnProgress) Handle(ctx context.Context, r slog.Record) error { r.Attrs(func(a slog.Attr) bool { if a.Key == "result" && a.Value.String() == "旧框架备份已校验" { h.once.Do(h.cancel) } return true }) return h.Handler.Handle(ctx, r) } func (h cancelOnProgress) WithAttrs(attrs []slog.Attr) slog.Handler { return cancelOnProgress{Handler: h.Handler.WithAttrs(attrs), cancel: h.cancel, once: h.once} } func (h cancelOnProgress) WithGroup(name string) slog.Handler { return cancelOnProgress{Handler: h.Handler.WithGroup(name), cancel: h.cancel, once: h.once} } func TestMySQLUpgradeResumesInterruptedFrameworkDDL(t *testing.T) { db := localDatabase(t) password := legacyFixture(t, db) root := t.TempDir() ctx, cancel := context.WithCancel(context.Background()) defer cancel() log := slog.New(cancelOnProgress{Handler: slog.NewTextHandler(io.Discard, nil), cancel: cancel, once: &sync.Once{}}) if err := swmigration.Run(ctx, db, modules.Catalog(), root, log); err == nil { t.Fatal("interrupted framework migration unexpectedly completed") } if err := swmigration.Run(context.Background(), db, modules.Catalog(), root, slog.New(slog.NewTextHandler(io.Discard, nil))); err != nil { t.Fatalf("recover interrupted DDL: %v", err) } assertComplete(t, db, password) } func TestMySQLUpgradeFinishesCommittedArchiveWithoutRemapping(t *testing.T) { db := localDatabase(t) password := legacyFixture(t, db) pool, _ := db.DB() conn, err := pool.Conn(context.Background()) if err != nil { t.Fatal(err) } statements, err := assets.Statements(assets.FrameworkSQL()) if err != nil { t.Fatal(err) } for _, statement := range statements { if strings.HasPrefix(strings.ToUpper(statement), "USE ") { continue } if _, err := conn.ExecContext(context.Background(), statement); err != nil { conn.Close() t.Fatal(err) } } conn.Close() if err := db.Exec("UPDATE sw_kra_state SET phase='data_validated',completed_at=NULL WHERE id=1").Error; err != nil { t.Fatal(err) } if err := swmigration.Run(context.Background(), db, modules.Catalog(), t.TempDir(), slog.New(slog.NewTextHandler(io.Discard, nil))); err != nil { t.Fatalf("resume committed conversion: %v", err) } assertComplete(t, db, password) } func TestMySQLNativeStartupAutomaticallyUpgradesWithLegacyDisableFlag(t *testing.T) { db := localDatabase(t) password := legacyFixture(t, db) cfg, err := driver.ParseDSN(os.Getenv("SW_TEST_MYSQL_DSN")) if err != nil { t.Fatal(err) } if err := db.Raw("SELECT DATABASE()").Scan(&cfg.DBName).Error; err != nil { t.Fatal(err) } cfg.ParseTime = true cfg.Loc = time.FixedZone("Asia/Shanghai", 8*3600) cfg.ReadTimeout = time.Second root := t.TempDir() path := filepath.Join(root, "config.yaml") if err := os.WriteFile(path, []byte("admin: {}\n"), 0600); err != nil { t.Fatal(err) } admin := &config.Admin{ConfigPath: path, System: &config.System{DisableAutoMigrate: true}, Local: &config.Local{StorePath: filepath.Join(root, "uploads")}} runtime := config.NewStore(&config.Config{Data: &config.Data{Database: &config.Database{Driver: "mysql", Source: cfg.FormatDSN(), LogMode: "silent", Engine: "InnoDB", MaxOpenConns: 1}}, Admin: admin}) defer runtime.Close() store, err := storage.NewReloadable(admin) if err != nil { t.Fatal(err) } d, closeData, err := rootdata.NewData(runtime, slog.New(slog.NewTextHandler(io.Discard, nil)), store, modules.Catalog()) if err != nil { t.Fatal(err) } defer closeData() if !d.DatabaseReady() { t.Fatal("database was not ready after automatic startup") } assertComplete(t, db, password) files, _ := filepath.Glob(filepath.Join(root, "migration-backups", "*.sql.gz")) if len(files) != 2 { t.Fatalf("automatic private backups not created: %v", files) } } func TestMySQLExistingMigratedInstallationPreservesTaskChoices(t *testing.T) { db := localDatabase(t) legacyFixture(t, db) pool, _ := db.DB() conn, err := pool.Conn(context.Background()) if err != nil { t.Fatal(err) } statements, err := assets.Statements(assets.FrameworkSQL()) if err != nil { t.Fatal(err) } for _, statement := range statements { if strings.HasPrefix(strings.ToUpper(statement), "USE ") { continue } if _, err := conn.ExecContext(context.Background(), statement); err != nil { conn.Close() t.Fatal(err) } } conn.Close() var oldSteps []migration.Step for _, step := range modules.Catalog().MigrationSteps() { if !strings.HasPrefix(step.ID, "202611060024_") && !strings.HasPrefix(step.ID, "202611060025_") { oldSteps = append(oldSteps, step) } } if err := migration.Run(db, oldSteps); err != nil { t.Fatal(err) } if err := db.Exec("UPDATE sys_timed_tasks SET enabled=(id=1)").Error; err != nil { t.Fatal(err) } var before, after []map[string]any if err := db.Table("sys_timed_tasks").Select("id,name,method_name,spec,params,enabled").Order("id").Find(&before).Error; err != nil { t.Fatal(err) } if err := swmigration.Run(context.Background(), db, modules.Catalog(), t.TempDir(), slog.New(slog.NewTextHandler(io.Discard, nil))); err != nil { t.Fatal(err) } if err := db.Table("sys_timed_tasks").Select("id,name,method_name,spec,params,enabled").Order("id").Find(&after).Error; err != nil { t.Fatal(err) } if !reflect.DeepEqual(before, after) { t.Fatal("automatic cleanup changed existing task configuration") } for _, name := range assets.CleanupTables() { if db.Migrator().HasTable(name) { t.Fatalf("cleanup omitted %s", name) } } }