// Package migrations 提供仅由命令行显式触发的版本化数据库迁移。 // Logan.Li package migrations import ( "embed" "errors" "fmt" "io/fs" "sort" "strings" "time" "ezops/model" "ezops/pkg/database" "gorm.io/gorm" ) const ( // BaselineVersion 标识当前模型集合的初始基线,不应在已有库上自动执行。 // Logan.Li BaselineVersion = "0000_baseline_schema" migrationTable = "sys_schema_migrations" ) //go:embed *.sql var migrationFiles embed.FS // Migration 是一个不可变、更名即视为兼容性破坏的数据库变更单元。 // Logan.Li type Migration struct { Version string Description string Apply func(*gorm.DB) error } // StatusRow 描述一个迁移版本的执行状态。 // Logan.Li type StatusRow struct { Version string Description string Applied bool AppliedAt *time.Time } // All 返回按版本排序的所有迁移。新迁移只能追加,禁止修改已发布版本。 // Logan.Li func All() []Migration { return []Migration{ { Version: BaselineVersion, Description: "新数据库初始结构基线", Apply: database.BootstrapSchema, }, { Version: "0001_agent_credentials", Description: "Agent 主机绑定凭证", Apply: func(db *gorm.DB) error { return executeDialectSQL(db, "0001_agent_credentials") }, }, { Version: "0002_legacy_schema_patches", Description: "历史库字段兼容补丁", Apply: database.ApplyLegacySchemaPatches, }, { Version: "0003_billing_settings", Description: "云账单汇率设置", Apply: func(db *gorm.DB) error { return executeDialectSQL(db, "0003_billing_settings") }, }, { Version: "0004_metric_aggregate_unique", Description: "指标聚合表唯一约束(upsert 生效)", Apply: func(db *gorm.DB) error { // 新库基线已由模型自带 uniq_metric_agg(uniqueIndex 标签)建出该索引, // 此时必须跳过:否则 CREATE UNIQUE INDEX 会以 1061 重复键名中断整个 migrate up, // 导致后面的迁移(如 0005_agent_settings)永远建不出来。 // 只有老库(历史 idx_agg 为非唯一索引)才需要去重 + 重建唯一索引。 // Logan.Li exists, err := database.IndexExists(db, "monitor_metric_aggregates", "uniq_metric_agg") if err != nil { return err } if exists { return nil } return executeDialectSQL(db, "0004_metric_aggregate_unique") }, }, { Version: "0005_agent_settings", Description: "Agent 采集策略(可下发频率)", Apply: func(db *gorm.DB) error { return executeDialectSQL(db, "0005_agent_settings") }, }, { Version: "0006_site_content_monitor", Description: "站点拨测内容监控(基线 + 变更)", Apply: func(db *gorm.DB) error { // 用 AutoMigrate 而非静态 SQL:它是幂等的。 // 空库上 0000_baseline_schema 已经建好这些列与表,静态 ALTER 会因 // duplicate column 失败;旧库则正好在这里补齐缺失的列/表。 // Logan.Li return db.AutoMigrate( &model.SiteTask{}, &model.SiteBaseline{}, &model.SiteContentChange{}, ) }, }, } } // Up 执行尚未应用的迁移。已有业务表但未完成基线登记时拒绝执行,防止误触发结构重建。 // Logan.Li func Up(db *gorm.DB) ([]string, error) { // 旧业务库缺少迁移记录表时必须零写入拒绝,避免误操作留下半初始化痕迹。 // Logan.Li if hasExistingApplicationSchema(db) && !db.Migrator().HasTable(migrationTable) { return nil, fmt.Errorf("检测到已有业务库但缺少 %s;请在备份与维护窗口确认后执行 `aiops migrate baseline`", migrationTable) } if err := ensureMigrationTable(db); err != nil { return nil, err } applied, err := appliedVersions(db) if err != nil { return nil, err } if _, baselineApplied := applied[BaselineVersion]; !baselineApplied && hasExistingApplicationSchema(db) { return nil, fmt.Errorf("检测到已有业务库但未登记 %s;请在备份与维护窗口确认后执行 `aiops migrate baseline`", BaselineVersion) } completed := make([]string, 0) for _, migration := range All() { if _, alreadyApplied := applied[migration.Version]; alreadyApplied { continue } if err := migration.Apply(db); err != nil { return completed, fmt.Errorf("迁移 %s(%s)失败: %w", migration.Version, migration.Description, err) } if err := recordApplied(db, migration.Version); err != nil { return completed, fmt.Errorf("登记迁移 %s 失败: %w", migration.Version, err) } completed = append(completed, migration.Version) } return completed, nil } // Baseline 仅为已经存在且已人工核验的旧数据库写入初始基线记录,不修改业务表。 // Logan.Li func Baseline(db *gorm.DB) error { if err := ensureMigrationTable(db); err != nil { return err } applied, err := appliedVersions(db) if err != nil { return err } if _, baselineApplied := applied[BaselineVersion]; baselineApplied { return nil } if !hasExistingApplicationSchema(db) { return errors.New("数据库为空,不能执行 baseline;请改用 `aiops migrate up`") } if err := validateLegacyBaseline(db); err != nil { return err } return recordApplied(db, BaselineVersion) } // Status 返回每个已知迁移的状态;只读查询,记录表不存在时全部显示 pending。 // Logan.Li func Status(db *gorm.DB) ([]StatusRow, error) { applied := map[string]time.Time{} if db.Migrator().HasTable(migrationTable) { var err error applied, err = appliedVersions(db) if err != nil { return nil, err } } rows := make([]StatusRow, 0, len(All())) for _, migration := range All() { row := StatusRow{ Version: migration.Version, Description: migration.Description, } if appliedAt, ok := applied[migration.Version]; ok { row.Applied = true row.AppliedAt = &appliedAt } rows = append(rows, row) } return rows, nil } func ensureMigrationTable(db *gorm.DB) error { sql := `CREATE TABLE IF NOT EXISTS sys_schema_migrations ( version VARCHAR(128) PRIMARY KEY, applied_at DATETIME NOT NULL )` if db.Dialector.Name() == "mysql" { sql = `CREATE TABLE IF NOT EXISTS sys_schema_migrations ( version VARCHAR(128) PRIMARY KEY, applied_at DATETIME NOT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4` } else if db.Dialector.Name() == "postgres" { sql = `CREATE TABLE IF NOT EXISTS sys_schema_migrations ( version VARCHAR(128) PRIMARY KEY, applied_at TIMESTAMP NOT NULL )` } if err := db.Exec(sql).Error; err != nil { return fmt.Errorf("创建迁移记录表失败: %w", err) } return nil } func appliedVersions(db *gorm.DB) (map[string]time.Time, error) { type appliedMigration struct { Version string AppliedAt time.Time } var records []appliedMigration if err := db.Table(migrationTable).Select("version, applied_at").Find(&records).Error; err != nil { return nil, fmt.Errorf("读取迁移记录失败: %w", err) } result := make(map[string]time.Time, len(records)) for _, record := range records { result[record.Version] = record.AppliedAt } return result, nil } func recordApplied(db *gorm.DB, version string) error { return db.Exec("INSERT INTO sys_schema_migrations (version, applied_at) VALUES (?, ?)", version, time.Now()).Error } func hasExistingApplicationSchema(db *gorm.DB) bool { return db.Migrator().HasTable("iam_users") || db.Migrator().HasTable("asset_servers") } func validateLegacyBaseline(db *gorm.DB) error { requiredTables := []string{"iam_users", "iam_roles", "asset_servers", "deploy_pipelines", "monitor_agent_reports"} for _, table := range requiredTables { if !db.Migrator().HasTable(table) { return fmt.Errorf("旧库缺少基线校验表 %s;请先恢复完整数据库或编写专项迁移", table) } } return nil } func executeDialectSQL(db *gorm.DB, prefix string) error { files, err := fs.Glob(migrationFiles, prefix+".*.sql") if err != nil { return fmt.Errorf("查找迁移脚本失败: %w", err) } sort.Strings(files) dialect := db.Dialector.Name() fileName := prefix + "." + dialect + ".sql" if !contains(files, fileName) { return fmt.Errorf("迁移 %s 不支持数据库驱动 %s", prefix, dialect) } content, err := migrationFiles.ReadFile(fileName) if err != nil { return fmt.Errorf("读取迁移脚本失败: %w", err) } statements := splitStatements(string(content)) for _, statement := range statements { if err := db.Exec(statement).Error; err != nil { return fmt.Errorf("执行 %s 失败: %w", fileName, err) } } return nil } func splitStatements(content string) []string { parts := strings.Split(content, ";") statements := make([]string, 0, len(parts)) for _, part := range parts { withoutComments := make([]string, 0) for _, line := range strings.Split(part, "\n") { line = strings.TrimSpace(line) if line != "" && !strings.HasPrefix(line, "--") { withoutComments = append(withoutComments, line) } } if statement := strings.TrimSpace(strings.Join(withoutComments, "\n")); statement != "" { statements = append(statements, statement) } } return statements } func contains(items []string, expected string) bool { for _, item := range items { if item == expected { return true } } return false }