package database import ( "errors" "fmt" "time" "gorm.io/gorm" ) const CurrentSchemaVersion int64 = 1 const baselineSchemaChecksum = "sha256:open-ai-canvas-schema-v1-20260830" const postgresSchemaMigrationLockID int64 = 73123910420260830 type SchemaStatus struct { Current int64 `json:"current"` Expected int64 `json:"expected"` Ready bool `json:"ready"` } type schemaMigration struct { Version int64 `gorm:"primaryKey"` Name string `gorm:"size:160;not null"` Checksum string `gorm:"size:96;not null"` AppliedAt time.Time `gorm:"not null"` } func (schemaMigration) TableName() string { return "schema_migrations" } type migration struct { version int64 name string checksum string apply func(*gorm.DB) error } var schemaMigrations = []migration{ {version: 1, name: "baseline_gorm_schema", checksum: baselineSchemaChecksum, apply: migrateSchemaV1}, } func MigrateSchema(db *gorm.DB) error { return db.Transaction(func(tx *gorm.DB) error { if tx.Dialector.Name() == "postgres" { if err := tx.Exec("SELECT pg_advisory_xact_lock(?)", postgresSchemaMigrationLockID).Error; err != nil { return fmt.Errorf("获取数据库迁移锁:%w", err) } } if err := tx.AutoMigrate(&schemaMigration{}); err != nil { return fmt.Errorf("初始化数据库迁移记录:%w", err) } for _, item := range schemaMigrations { var applied schemaMigration err := tx.First(&applied, "version = ?", item.version).Error if err == nil { if err := validateMigrationRecord(applied, item); err != nil { return err } continue } if !errors.Is(err, gorm.ErrRecordNotFound) { return fmt.Errorf("读取数据库迁移 %d:%w", item.version, err) } if err := item.apply(tx); err != nil { return fmt.Errorf("执行数据库迁移 %d(%s):%w", item.version, item.name, err) } record := schemaMigration{Version: item.version, Name: item.name, Checksum: item.checksum, AppliedAt: time.Now().UTC()} if err := tx.Create(&record).Error; err != nil { return fmt.Errorf("记录数据库迁移 %d:%w", item.version, err) } } return RequireSchemaVersion(tx) }) } func ReadSchemaStatus(db *gorm.DB) (SchemaStatus, error) { status := SchemaStatus{Expected: CurrentSchemaVersion} if !db.Migrator().HasTable(&schemaMigration{}) { return status, nil } if err := db.Model(&schemaMigration{}).Select("COALESCE(MAX(version), 0)").Scan(&status.Current).Error; err != nil { return status, fmt.Errorf("读取数据库结构版本:%w", err) } if status.Current != status.Expected { return status, nil } if err := validateMigrationRecords(db); err != nil { return status, err } status.Ready = true return status, nil } func validateMigrationRecords(db *gorm.DB) error { for _, item := range schemaMigrations { var applied schemaMigration if err := db.First(&applied, "version = ?", item.version).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return fmt.Errorf("数据库缺少迁移记录 %d(%s)", item.version, item.name) } return fmt.Errorf("读取数据库迁移 %d:%w", item.version, err) } if err := validateMigrationRecord(applied, item); err != nil { return err } } return nil } func validateMigrationRecord(applied schemaMigration, expected migration) error { if applied.Name != expected.name { return fmt.Errorf("数据库迁移 %d 名称不一致:记录为 %s,程序期望 %s", expected.version, applied.Name, expected.name) } if applied.Checksum != expected.checksum { return fmt.Errorf("数据库迁移 %d 校验和不一致:记录为 %s,程序期望 %s", expected.version, applied.Checksum, expected.checksum) } return nil } func RequireSchemaVersion(db *gorm.DB) error { status, err := ReadSchemaStatus(db) if err != nil { return err } if status.Current < status.Expected { return fmt.Errorf("数据库结构版本过旧:当前 %d,程序要求 %d,请先执行 migrate-schema up", status.Current, status.Expected) } if status.Current > status.Expected { return fmt.Errorf("数据库结构版本 %d 高于程序支持的 %d,拒绝使用旧程序连接新数据库", status.Current, status.Expected) } return nil }