216 lines
8.7 KiB
Go
216 lines
8.7 KiB
Go
package database
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"fmt"
|
||
"reflect"
|
||
"regexp"
|
||
"sort"
|
||
"strings"
|
||
)
|
||
|
||
// MySQL 的建表语句会单独提交，恢复时必须校验结构，不能直接吞掉同名表错误。
|
||
func applyMySQLDeploymentMigration(ctx context.Context, db *sql.DB, version int, statements []string) error {
|
||
if len(statements) == 0 {
|
||
return fmt.Errorf("部署迁移缺少建表语句：%d", version)
|
||
}
|
||
shape, err := parseMySQLDeploymentShape(statements[0])
|
||
if err != nil {
|
||
return err
|
||
}
|
||
create := strings.Replace(statements[0], "CREATE TABLE ", "CREATE TABLE IF NOT EXISTS ", 1)
|
||
if _, err := db.ExecContext(ctx, create); err != nil {
|
||
return fmt.Errorf("创建部署迁移表失败：%w", err)
|
||
}
|
||
if err := validateMySQLDeploymentShape(ctx, db, shape); err != nil {
|
||
return err
|
||
}
|
||
tx, err := db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer tx.Rollback()
|
||
if version == 15 {
|
||
if err := backfillMySQLTaskIdempotency(ctx, tx); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if _, err := tx.ExecContext(ctx, "INSERT INTO schema_migrations (version) VALUES (?)", version); err != nil {
|
||
return fmt.Errorf("记录部署迁移版本失败：%w", err)
|
||
}
|
||
return tx.Commit()
|
||
}
|
||
|
||
type mysqlDeploymentColumn struct {
|
||
Name string
|
||
Type string
|
||
Default string
|
||
HasDefault bool
|
||
}
|
||
|
||
type mysqlDeploymentShape struct {
|
||
Table string
|
||
Columns []mysqlDeploymentColumn
|
||
Indexes []string
|
||
}
|
||
|
||
var (
|
||
mysqlDeploymentTablePattern = regexp.MustCompile(`^CREATE TABLE ([a-z_]+) \($`)
|
||
mysqlDeploymentColumnPattern = regexp.MustCompile(`^([a-z_]+) (VARCHAR\([0-9]+\)|INT|TEXT|MEDIUMTEXT|DATETIME) NOT NULL(?: DEFAULT (CURRENT_TIMESTAMP|0))?( PRIMARY KEY)?$`)
|
||
mysqlDeploymentIndexPattern = regexp.MustCompile(`^(PRIMARY KEY|UNIQUE KEY ([a-z_]+)|INDEX ([a-z_]+)) \(([a-z_, ]+)\)$`)
|
||
mysqlIntegerWidthPattern = regexp.MustCompile(`^int\([0-9]+\)$`)
|
||
)
|
||
|
||
// 只解析部署与报告内嵌迁移使用的语法；出现未知定义时拒绝迁移，避免漏检结构。
|
||
func parseMySQLDeploymentShape(statement string) (mysqlDeploymentShape, error) {
|
||
var shape mysqlDeploymentShape
|
||
lines := strings.Split(strings.TrimSpace(statement), "\n")
|
||
if len(lines) < 3 {
|
||
return shape, fmt.Errorf("部署迁移表定义不完整")
|
||
}
|
||
match := mysqlDeploymentTablePattern.FindStringSubmatch(strings.TrimSpace(lines[0]))
|
||
if len(match) != 2 || strings.TrimSpace(lines[len(lines)-1]) != ")" {
|
||
return shape, fmt.Errorf("部署迁移表定义不受支持")
|
||
}
|
||
shape.Table = match[1]
|
||
for _, raw := range lines[1 : len(lines)-1] {
|
||
line := strings.TrimSuffix(strings.TrimSpace(raw), ",")
|
||
if parts := mysqlDeploymentColumnPattern.FindStringSubmatch(line); len(parts) == 5 {
|
||
shape.Columns = append(shape.Columns, mysqlDeploymentColumn{parts[1], strings.ToLower(parts[2]), strings.ToLower(parts[3]), parts[3] != ""})
|
||
if parts[4] != "" {
|
||
shape.Indexes = append(shape.Indexes, "PRIMARY:0:1:"+parts[1])
|
||
}
|
||
continue
|
||
}
|
||
parts := mysqlDeploymentIndexPattern.FindStringSubmatch(line)
|
||
if len(parts) != 5 {
|
||
return shape, fmt.Errorf("部署迁移表包含不受支持的定义：%s", shape.Table)
|
||
}
|
||
name, nonUnique := "PRIMARY", 0
|
||
if parts[2] != "" {
|
||
name = parts[2]
|
||
} else if parts[3] != "" {
|
||
name, nonUnique = parts[3], 1
|
||
}
|
||
for index, column := range strings.Split(parts[4], ",") {
|
||
shape.Indexes = append(shape.Indexes, fmt.Sprintf("%s:%d:%d:%s", name, nonUnique, index+1, strings.TrimSpace(column)))
|
||
}
|
||
}
|
||
sort.Strings(shape.Indexes)
|
||
return shape, nil
|
||
}
|
||
|
||
func validateMySQLDeploymentShape(ctx context.Context, db *sql.DB, expected mysqlDeploymentShape) error {
|
||
var tableCollation, schemaCollation string
|
||
if err := db.QueryRowContext(ctx, `SELECT t.table_collation, s.default_collation_name
|
||
FROM information_schema.tables t JOIN information_schema.schemata s ON t.table_schema = s.schema_name
|
||
WHERE t.table_schema = DATABASE() AND t.table_name = ?`, expected.Table).Scan(&tableCollation, &schemaCollation); err != nil {
|
||
return err
|
||
}
|
||
if tableCollation != schemaCollation {
|
||
return fmt.Errorf("部署迁移表排序规则与数据库默认定义不匹配：%s", expected.Table)
|
||
}
|
||
var engine, tableType string
|
||
if err := db.QueryRowContext(ctx, `SELECT ENGINE, TABLE_TYPE FROM information_schema.tables
|
||
WHERE table_schema = DATABASE() AND table_name = ?`, expected.Table).Scan(&engine, &tableType); err != nil {
|
||
return fmt.Errorf("读取部署迁移表状态失败：%w", err)
|
||
}
|
||
if !strings.EqualFold(engine, "InnoDB") || tableType != "BASE TABLE" {
|
||
return fmt.Errorf("部署迁移表必须是 InnoDB 基础表：%s", expected.Table)
|
||
}
|
||
rows, err := db.QueryContext(ctx, `SELECT column_name, column_type, column_default, is_nullable, extra, collation_name
|
||
FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = ? ORDER BY ordinal_position`, expected.Table)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
var actual []mysqlDeploymentColumn
|
||
for rows.Next() {
|
||
var column mysqlDeploymentColumn
|
||
var defaultValue sql.NullString
|
||
var columnCollation sql.NullString
|
||
var nullable, extra string
|
||
if err := rows.Scan(&column.Name, &column.Type, &defaultValue, &nullable, &extra, &columnCollation); err != nil {
|
||
rows.Close()
|
||
return err
|
||
}
|
||
column.Type = strings.ToLower(column.Type)
|
||
if mysqlIntegerWidthPattern.MatchString(column.Type) {
|
||
column.Type = "int"
|
||
}
|
||
column.Default = strings.TrimSuffix(strings.ToLower(defaultValue.String), "()")
|
||
column.HasDefault = defaultValue.Valid
|
||
if columnCollation.Valid && columnCollation.String != tableCollation {
|
||
rows.Close()
|
||
return fmt.Errorf("部署迁移表字段排序规则不匹配：%s.%s", expected.Table, column.Name)
|
||
}
|
||
if nullable != "NO" || (extra != "" && !(strings.EqualFold(extra, "DEFAULT_GENERATED") && column.Default == "current_timestamp")) {
|
||
rows.Close()
|
||
return fmt.Errorf("部署迁移表字段属性不匹配：%s.%s", expected.Table, column.Name)
|
||
}
|
||
actual = append(actual, column)
|
||
}
|
||
err = rows.Err()
|
||
rows.Close()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !reflect.DeepEqual(actual, expected.Columns) {
|
||
return fmt.Errorf("部署迁移表字段定义不匹配：%s，拒绝跳过或覆盖现有数据", expected.Table)
|
||
}
|
||
rows, err = db.QueryContext(ctx, `SELECT index_name, non_unique, seq_in_index, column_name, sub_part, index_type, collation
|
||
FROM information_schema.statistics WHERE table_schema = DATABASE() AND table_name = ?`, expected.Table)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
var indexes []string
|
||
for rows.Next() {
|
||
var name, column, indexType, collation string
|
||
var nonUnique, position int
|
||
var prefix sql.NullInt64
|
||
if err := rows.Scan(&name, &nonUnique, &position, &column, &prefix, &indexType, &collation); err != nil {
|
||
rows.Close()
|
||
return err
|
||
}
|
||
if prefix.Valid || indexType != "BTREE" || collation != "A" {
|
||
rows.Close()
|
||
return fmt.Errorf("部署迁移表索引属性不匹配：%s.%s", expected.Table, name)
|
||
}
|
||
indexes = append(indexes, fmt.Sprintf("%s:%d:%d:%s", name, nonUnique, position, column))
|
||
}
|
||
err = rows.Err()
|
||
rows.Close()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
sort.Strings(indexes)
|
||
if !reflect.DeepEqual(indexes, expected.Indexes) {
|
||
return fmt.Errorf("部署迁移表索引定义不匹配：%s，拒绝跳过或覆盖现有数据", expected.Table)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// 回填与版本标记同事务提交；已落盘的历史回填只有内容完全一致才可复用。
|
||
func backfillMySQLTaskIdempotency(ctx context.Context, tx *sql.Tx) error {
|
||
const expected = `SELECT domain_id, idempotency_key, MIN(id) AS task_id, MIN(created_at) AS created_at
|
||
FROM tasks WHERE idempotency_key <> '' GROUP BY domain_id, idempotency_key`
|
||
var conflicts int
|
||
if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM task_idempotency AS actual
|
||
LEFT JOIN (`+expected+`) AS expected ON actual.domain_id = expected.domain_id AND actual.idempotency_key = expected.idempotency_key
|
||
WHERE expected.task_id IS NULL OR BINARY actual.task_id <> BINARY expected.task_id OR actual.created_at <> expected.created_at
|
||
OR BINARY actual.domain_id <> BINARY expected.domain_id OR BINARY actual.idempotency_key <> BINARY expected.idempotency_key`).Scan(&conflicts); err != nil {
|
||
return fmt.Errorf("检查任务幂等历史回填失败：%w", err)
|
||
}
|
||
if conflicts != 0 {
|
||
return fmt.Errorf("任务幂等历史回填与已有数据冲突，拒绝覆盖，需核查后重试")
|
||
}
|
||
_, err := tx.ExecContext(ctx, `INSERT INTO task_idempotency (domain_id, idempotency_key, task_id, created_at)
|
||
SELECT expected.domain_id, expected.idempotency_key, expected.task_id, expected.created_at FROM (`+expected+`) AS expected
|
||
LEFT JOIN task_idempotency AS actual ON actual.domain_id = expected.domain_id AND actual.idempotency_key = expected.idempotency_key
|
||
WHERE actual.domain_id IS NULL`)
|
||
if err != nil {
|
||
return fmt.Errorf("回填任务幂等历史失败：%w", err)
|
||
}
|
||
return nil
|
||
}
|