257 lines
7.7 KiB
Go
257 lines
7.7 KiB
Go
// Package migrations 提供仅由命令行显式触发的版本化数据库迁移。 // Logan.Li
|
||
package migrations
|
||
|
||
import (
|
||
"embed"
|
||
"errors"
|
||
"fmt"
|
||
"io/fs"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
"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")
|
||
},
|
||
},
|
||
}
|
||
}
|
||
|
||
// 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
|
||
}
|