aiops/backend/migrations/migrations.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")
			},
		},
		{
			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")
			},
		},
	}
}

// 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
}