实现多商户履约平台基础

2026-07-30 12:02:32 +08:00
parent dde6f2c269
commit 83429456a6
41 changed files with 5487 additions and 504 deletions
@@ -0,0 +1,168 @@
+package database
+
+import (
+	"crypto/sha256"
+	"embed"
+	"encoding/hex"
+	"fmt"
+	"io/fs"
+	"path/filepath"
+	"sort"
+	"strings"
+	"time"
+
+	"affiliate_dash/internal/model"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+//go:embed migrations/*.sql
+var migrationFiles embed.FS
+
+type schemaMigration struct {
+	Version   string    `gorm:"primaryKey;size:64"`
+	Name      string    `gorm:"size:255;not null"`
+	Checksum  string    `gorm:"size:64;not null"`
+	AppliedAt time.Time `gorm:"not null"`
+}
+
+func (schemaMigration) TableName() string {
+	return "schema_migrations"
+}
+
+// Migrate 使用显式 SQL 迁移管理表结构，避免大型项目依赖隐式结构同步。
+func Migrate(db *gorm.DB) error {
+	return db.Transaction(func(tx *gorm.DB) error {
+		if err := tx.Exec("SELECT pg_advisory_xact_lock(hashtext(?))", "affiliate_dash_schema_migrations").Error; err != nil {
+			return fmt.Errorf("lock schema migrations: %w", err)
+		}
+		if err := ensureMigrationTable(tx); err != nil {
+			return err
+		}
+		if err := applySQLMigrations(tx); err != nil {
+			return err
+		}
+		merchant, err := ensureSelfMerchant(tx)
+		if err != nil {
+			return err
+		}
+		return ensureWallet(tx, merchant.ID)
+	})
+}
+
+func ensureMigrationTable(tx *gorm.DB) error {
+	statements := []string{
+		`
+CREATE TABLE IF NOT EXISTS schema_migrations (
+	version varchar(64) PRIMARY KEY,
+	name varchar(255) NOT NULL,
+	checksum varchar(64) NOT NULL,
+	applied_at timestamptz NOT NULL
+)`,
+		`ALTER TABLE schema_migrations ADD COLUMN IF NOT EXISTS name varchar(255) NOT NULL DEFAULT ''`,
+		`ALTER TABLE schema_migrations ADD COLUMN IF NOT EXISTS checksum varchar(64) NOT NULL DEFAULT ''`,
+		`ALTER TABLE schema_migrations ADD COLUMN IF NOT EXISTS applied_at timestamptz NOT NULL DEFAULT now()`,
+	}
+	for _, statement := range statements {
+		if err := tx.Exec(statement).Error; err != nil {
+			return fmt.Errorf("ensure schema_migrations: %w", err)
+		}
+	}
+	return nil
+}
+
+func applySQLMigrations(tx *gorm.DB) error {
+	paths, err := fs.Glob(migrationFiles, "migrations/*.sql")
+	if err != nil {
+		return fmt.Errorf("list migrations: %w", err)
+	}
+	sort.Strings(paths)
+	for _, path := range paths {
+		content, err := migrationFiles.ReadFile(path)
+		if err != nil {
+			return fmt.Errorf("read migration %s: %w", path, err)
+		}
+		version, name := parseMigrationName(path)
+		checksum := migrationChecksum(content)
+
+		var existing schemaMigration
+		err = tx.Where("version = ?", version).First(&existing).Error
+		if err == nil {
+			if existing.Checksum == "" {
+				if err := tx.Model(&schemaMigration{}).Where("version = ?", version).Updates(map[string]interface{}{
+					"name":     name,
+					"checksum": checksum,
+				}).Error; err != nil {
+					return fmt.Errorf("adopt migration %s: %w", version, err)
+				}
+				continue
+			}
+			if existing.Checksum != checksum {
+				return fmt.Errorf("migration %s checksum mismatch: 已执行版本不能被修改", version)
+			}
+			continue
+		}
+		if err != gorm.ErrRecordNotFound {
+			return fmt.Errorf("query migration %s: %w", version, err)
+		}
+		sql := strings.TrimSpace(string(content))
+		if sql == "" {
+			return fmt.Errorf("migration %s is empty", path)
+		}
+		if err := tx.Exec(sql).Error; err != nil {
+			return fmt.Errorf("apply migration %s: %w", path, err)
+		}
+		if err := tx.Create(&schemaMigration{
+			Version:   version,
+			Name:      name,
+			Checksum:  checksum,
+			AppliedAt: time.Now(),
+		}).Error; err != nil {
+			return fmt.Errorf("record migration %s: %w", version, err)
+		}
+	}
+	return nil
+}
+
+func parseMigrationName(path string) (string, string) {
+	base := filepath.Base(path)
+	name := strings.TrimSuffix(base, filepath.Ext(base))
+	parts := strings.SplitN(name, "_", 2)
+	if len(parts) == 1 {
+		return parts[0], name
+	}
+	return parts[0], parts[1]
+}
+
+func migrationChecksum(content []byte) string {
+	sum := sha256.Sum256(content)
+	return hex.EncodeToString(sum[:])
+}
+
+func ensureSelfMerchant(tx *gorm.DB) (*model.Merchant, error) {
+	var merchant model.Merchant
+	err := tx.Where("code = ?", "self-operated").First(&merchant).Error
+	if err == nil {
+		return &merchant, nil
+	}
+	if err != gorm.ErrRecordNotFound {
+		return nil, err
+	}
+	merchant = model.Merchant{
+		Code:   "self-operated",
+		Name:   "自营商户",
+		Status: model.MerchantStatusActive,
+	}
+	if err := tx.Create(&merchant).Error; err != nil {
+		return nil, err
+	}
+	return &merchant, nil
+}
+
+func ensureWallet(tx *gorm.DB, merchantID uint) error {
+	return tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&model.WalletAccount{
+		MerchantID: merchantID,
+		Currency:   "CNY",
+	}).Error
+}