第11章 · 不停机数据迁移（正式版）

// webook/internal/migration/init.go（异构初始化示意）
func MigrateData(ctx context.Context, srcDB, dstDB *gorm.DB) error {
	var srcRecords []SrcRecord
	// 步骤 1：分批读取源库数据，每批 500 条，避免一次加载过多
	err := srcDB.WithContext(ctx).Model(&SrcRecord{}).
		FindInBatches(&srcRecords, 500, func(tx *gorm.DB, batch int) error {
			// 步骤 2：逐条转换为目标库格式（类型/字段映射都在这里）
			for _, src := range srcRecords {
				dst := convertToDst(src)
				// 步骤 3：写入目标库
				if err := dstDB.Create(&dst).Error; err != nil {
					return err
				}
			}
			return nil
		}).Error
	return err
}

// webook/internal/migration/validator.go
package migration

import "gorm.io/gorm"

// Entity 是校验对象的接口：每张要校验的表，都实现它
type Entity interface {
	// ID 主键，整个校验体系围绕 ID 进行
	ID() int64
	// TableName 表名，修复时需要拼 SQL
	TableName() string
	// CompareTo 比较两个实体是否一致
	// 由实现者负责：明确知道每个字段怎么比较、要不要忽略某些列（如 utime）
	CompareTo(other Entity) error
	// Columns 所有列名，修复时用
	Columns() []string
}

// Validator 泛型校验器
type Validator[T Entity] struct {
	baseDB    *gorm.DB // 基准库（第二阶段=源表，第三阶段=目标库）
	targetDB  *gorm.DB // 对比库
	direction Direction // SRC 或 DST
}

// Direction 数据方向：以谁为准
type Direction uint8

const (
	DirectionSRC Direction = iota // 以源表为准
	DirectionDST                  // 以目标表为准
)

// Validate 执行校验，offset 是从哪个 ID 开始（游标分页）
func (v *Validator[T]) Validate(ctx context.Context, offset int64) (int64, error) {
	var base T
	// 步骤 1：从 base 中取一条数据（ID 大于 offset，保证按顺序推进）
	err := v.baseDB.WithContext(ctx).
		Where("id > ?", offset).
		First(&base).Error
	if err == gorm.ErrRecordNotFound {
		// 校验完毕：base 里没有更大的 ID 了
		return offset, nil
	}
	if err != nil {
		return offset, err
	}

	// 步骤 2：从 target 中按主键找对应数据
	var target T
	err = v.targetDB.WithContext(ctx).
		Where("id = ?", base.ID()).First(&target).Error
	if err == gorm.ErrRecordNotFound {
		// target 中没有，需要修复（insert）
		v.notifyFix(base.ID(), "target_missing")
		return base.ID(), nil
	}
	if err != nil {
		log.Warn("查询 target 出错", err)
		return base.ID(), nil
	}

	// 步骤 3：比较数据
	if cmpErr := base.CompareTo(target); cmpErr != nil {
		// 数据不一致，需要修复（update）
		v.notifyFix(base.ID(), "neq")
	}
	return base.ID(), nil
}

// ReverseValidate 反向校验：找 target 中有但 base 中没有的记录
func (v *Validator[T]) ReverseValidate(ctx context.Context) error {
	var ids []int64
	cursor := int64(0)
	for {
		ids = ids[:0]
		// 批量取 target 中的 ID，避免一条条查
		err := v.targetDB.WithContext(ctx).
			Where("id > ?", cursor).
			Limit(500).Pluck("id", &ids).Error
		if err != nil || len(ids) == 0 {
			return err
		}

		// 在 base 中找这些 ID
		var found []int64
		v.baseDB.WithContext(ctx).
			Where("id IN ?", ids).Pluck("id", &found).Error
		foundSet := make(map[int64]struct{}, len(found))
		for _, id := range found {
			foundSet[id] = struct{}{}
		}

		// 找出 target 有但 base 没有的，需要修复（delete from target）
		for _, id := range ids {
			if _, ok := foundSet[id]; !ok {
				v.notifyFix(id, "base_missing")
			}
		}
		cursor = ids[len(ids)-1]
	}
}

type FixMsg struct {
	ID        int64     `json:"id"`        // 主键
	Type      FixType   `json:"type"`      // 不一致类型：target_missing / neq / base_missing
	Direction Direction `json:"direction"` // 以 SRC 为准还是 DST 为准
	Table     string    `json:"table"`     // 表名
}

type FixType uint8

const (
	FixTypeTargetMissing FixType = iota // 目标表缺数据，执行 insert
	FixTypeNeq                          // 数据不相等，执行 update
	FixTypeBaseMissing                  // 目标表多数据，执行 delete
)

// Fix 修复一条数据
func (f *Fixer[T]) Fix(ctx context.Context, msg FixMsg) error {
	var base T
	// 再次查找 base 中的最新数据（避免使用校验时的旧快照）
	err := f.baseDB.WithContext(ctx).
		Where("id = ?", msg.ID).First(&base).Error
	if err == gorm.ErrRecordNotFound {
		// base 中没有了，删除 target 中的记录
		return f.targetDB.WithContext(ctx).
			Where("id = ?", msg.ID).Delete(base).Error
	}
	if err != nil {
		return err
	}
	// base 中有，执行 UPSERT（存在则更新，不存在则插入）
	return f.targetDB.WithContext(ctx).Save(&base).Error
}

// SimplifiedFix 简化版修复
func (f *Fixer[T]) SimplifiedFix(ctx context.Context, id int64) error {
	var base T
	err := f.baseDB.WithContext(ctx).Where("id = ?", id).First(&base).Error
	if err == gorm.ErrRecordNotFound {
		return f.targetDB.WithContext(ctx).Where("id = ?", id).Delete(base).Error
	}
	if err != nil {
		return err
	}
	return f.targetDB.WithContext(ctx).Save(&base).Error
}

// gorm.io/gorm 中的底层接口
type ConnPool interface {
	PrepareContext(ctx context.Context, query string) (*sql.Stmt, error)
	ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
	QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
	QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row
}

// webook/internal/migration/double_write_pool.go
package migration

import (
	"context"
	"database/sql"
	"fmt"
	"sync/atomic"

	"gorm.io/gorm"
)

// Pattern 双写模式
type Pattern uint8

const (
	PatternSRCOnly   Pattern = iota // 只读写源库
	PatternSRCFirst                 // 双写，读源库，源库为准
	PatternDSTFirst                 // 双写，读目标库，目标库为准
	PatternDSTOnly                  // 只读写目标库
)

// DoubleWritePool 双写连接池：替换 GORM 的 ConnPool
type DoubleWritePool struct {
	src     *gorm.DB    // 源库
	dst     *gorm.DB    // 目标库
	pattern atomic.Int32 // 当前模式，用原子变量保证并发安全切换
}

func NewDoubleWritePool(src, dst *gorm.DB) *DoubleWritePool {
	p := &DoubleWritePool{src: src, dst: dst}
	p.pattern.Store(int32(PatternSRCOnly))
	return p
}

// UpdatePattern 切换模式（运维通过 HTTP 接口调用它完成阶段切换）
func (p *DoubleWritePool) UpdatePattern(pat Pattern) {
	p.pattern.Store(int32(pat))
}

// ExecContext 处理增删改（写）
func (p *DoubleWritePool) ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error) {
	pat := Pattern(p.pattern.Load())
	switch pat {
	case PatternSRCOnly:
		return p.src.ExecContext(ctx, query, args...)
	case PatternDSTOnly:
		return p.dst.ExecContext(ctx, query, args...)
	case PatternSRCFirst:
		// 双写：基准是老库，先写 src（失败报错），再写 dst（失败吞错）
		res, err := p.src.ExecContext(ctx, query, args...)
		if err != nil {
			return res, err
		}
		if _, e := p.dst.ExecContext(ctx, query, args...); e != nil {
			log.Warn("双写 dst 失败，等校验修复兜底", e)
		}
		return res, nil
	case PatternDSTFirst:
		// 双写：基准是新库，先写 dst（失败报错），再写 src（失败吞错）
		res, err := p.dst.ExecContext(ctx, query, args...)
		if err != nil {
			return res, err
		}
		if _, e := p.src.ExecContext(ctx, query, args...); e != nil {
			log.Warn("双写 src 失败，等校验修复兜底", e)
		}
		return res, nil
	}
	return nil, fmt.Errorf("unknown pattern: %d", pat)
}

// QueryContext 处理查询（读，返回多行）
func (p *DoubleWritePool) QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error) {
	pat := Pattern(p.pattern.Load())
	switch pat {
	case PatternSRCOnly, PatternSRCFirst:
		return p.src.QueryContext(ctx, query, args...)
	case PatternDSTOnly, PatternDSTFirst:
		return p.dst.QueryContext(ctx, query, args...)
	}
	return nil, fmt.Errorf("unknown pattern: %d", pat)
}

// QueryRowContext 单行查询
func (p *DoubleWritePool) QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row {
	pat := Pattern(p.pattern.Load())
	switch pat {
	case PatternSRCOnly, PatternSRCFirst:
		return p.src.QueryRowContext(ctx, query, args...)
	case PatternDSTOnly, PatternDSTFirst:
		return p.dst.QueryRowContext(ctx, query, args...)
	}
	return p.src.QueryRowContext(ctx, query, args...)
}

// DoubleWriteTx 双写事务
type DoubleWriteTx struct {
	srcTx    *gorm.DB // 源库事务
	dstTx    *gorm.DB // 目标库事务
	pattern  Pattern
}

func (t *DoubleWriteTx) Commit() error {
	switch t.pattern {
	case PatternDSTOnly:
		// 只写目标库：只提交 dst
		return t.dstTx.Commit().Error
	case PatternDSTFirst:
		// DST_FIRST：目标库为基准，先提交 dst（失败报错），src 失败吞错
		if err := t.dstTx.Commit().Error; err != nil {
			return err
		}
		if err := t.srcTx.Commit().Error; err != nil {
			log.Warn("src 事务提交失败，等校验修复兜底", err)
		}
		return nil
	case PatternSRCFirst:
		// SRC_FIRST 双写：源库为基准，先提交 src（失败报错），dst 失败吞错
		if err := t.srcTx.Commit().Error; err != nil {
			return err
		}
		if err := t.dstTx.Commit().Error; err != nil {
			log.Warn("dst 事务提交失败，等校验修复兜底", err)
		}
		return nil
	default:
		// SRC_ONLY：只写源库，只提交 src
		return t.srcTx.Commit().Error
	}
}

func (t *DoubleWriteTx) Rollback() error {
	switch t.pattern {
	case PatternDSTOnly:
		// 只写目标库：只回滚 dst
		return t.dstTx.Rollback().Error
	case PatternDSTFirst:
		// DST_FIRST：目标库为基准，先回滚 dst，src 失败吞错
		if err := t.dstTx.Rollback().Error; err != nil {
			return err
		}
		if err := t.srcTx.Rollback().Error; err != nil {
			log.Warn("src 事务回滚失败", err)
		}
		return nil
	case PatternSRCFirst:
		// SRC_FIRST：先回滚 src，dst 失败吞错
		if err := t.srcTx.Rollback().Error; err != nil {
			return err
		}
		if err := t.dstTx.Rollback().Error; err != nil {
			log.Warn("dst 事务回滚失败", err)
		}
		return nil
	default:
		// SRC_ONLY：只回滚 src
		return t.srcTx.Rollback().Error
	}
}

// IncrementalValidate 增量校验：只校验 utime > since 的数据
func (v *Validator[T]) IncrementalValidate(ctx context.Context, since int64, sleepInterval time.Duration) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
		}
		var records []T
		// 步骤 1：取 utime 在 since 之后的数据（近期被修改的）
		err := v.baseDB.WithContext(ctx).
			Where("utime > ?", since).
			FindInBatches(&records, 500, func(tx *gorm.DB, batch int) error {
				// 步骤 2：对已取出的记录直接比对（不要再借 Validate(ctx, r.ID()-1) 重新查库——
				// 那样既浪费已取回的数据，又可能因为无 Order 查到别的行）
				for _, r := range records {
					var target T
					err := v.targetDB.WithContext(ctx).Where("id = ?", r.ID()).First(&target).Error
					if err == gorm.ErrRecordNotFound {
						v.notifyFix(r.ID(), "target_missing")
						continue
					}
					if err != nil {
						log.Warn("查询 target 出错", err)
						continue
					}
					if cmpErr := r.CompareTo(target); cmpErr != nil {
						v.notifyFix(r.ID(), "neq")
					}
				}
				return nil
			}).Error
		if err != nil {
			log.Warn("增量校验出错", err)
		}
		// sleepInterval ≤ 0 表示"跑完一轮全量即退出"
		if sleepInterval <= 0 {
			return
		}
		// 步骤 3：更新 since 为当前时间，下一轮只校验更新的
		// 注意：全书 utime 统一存毫秒时间戳，这里必须用 UnixMilli 保持单位一致
		since = time.Now().UnixMilli()
		time.Sleep(sleepInterval)
	}
}

// 形态一：utime=0, sleep=0 → 只跑一次全量校验后退出
go v.IncrementalValidate(ctx, 0, 0)

// 形态二：utime=0, sleep>0 → 先全量，结束后转持续增量
go v.IncrementalValidate(ctx, 0, 30*time.Second)

// 形态三：utime=近期, sleep>0 → 只跑增量，长期驻留（utime 是毫秒，1 小时 = 3600*1000）
go v.IncrementalValidate(ctx, time.Now().UnixMilli()-3600*1000, 30*time.Second)

// webook/internal/migration/scheduler.go（示意）
type Scheduler struct {
	validator *Validator[SomeEntity]
	fixer     *Fixer[SomeEntity]
	pool      *DoubleWritePool
}

func (s *Scheduler) FullValidate()                 { go s.validator.FullValidate(context.Background()) }
func (s *Scheduler) IncrementalValidate()          { go s.validator.IncrementalValidate(context.Background(), time.Now().UnixMilli(), 30*time.Second) }
func (s *Scheduler) SwitchPattern(pat Pattern)     { s.pool.UpdatePattern(pat) }

POST /migration/full-validate                     # 触发全量校验
POST /migration/incremental-validate              # 触发增量校验
POST /migration/switch-pattern?pattern=src_first  # 切换阶段
GET  /migration/status                            # 查看当前阶段与不一致计数

# 切到双写第二阶段（读源库、源为准）
curl -X POST "http://localhost:8082/migration/switch-pattern?pattern=src_first"
# 触发一次全量校验
curl -X POST "http://localhost:8082/migration/full-validate"

-- schema/user_service.sql：目标库建表脚本
CREATE DATABASE IF NOT EXISTS user_service DEFAULT CHARSET utf8mb4;
USE user_service;
CREATE TABLE IF NOT EXISTS `user` (
    `id`       BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    `email`    VARCHAR(255) NOT NULL DEFAULT '',
    `nickname` VARCHAR(64)  NOT NULL DEFAULT '',
    `password` VARCHAR(255) NOT NULL DEFAULT '',
    `ctime`    BIGINT NOT NULL DEFAULT 0,
    `utime`    BIGINT NOT NULL DEFAULT 0,
    UNIQUE KEY `uk_email` (`email`)
) ENGINE=InnoDB;

# configs/config.yaml：新增 mysql_new 与 migration 两段（其余保持不变）
mysql_new:
  dsn: "root:root@tcp(127.0.0.1:3306)/user_service?charset=utf8mb4&parseTime=True&loc=Local"
migration:
  double_write: true       # 双写开关，回滚时改为 false 并重启
  read_new_ratio: 0.01     # 灰度读比例：1% 的用户读新库（兜底值）
  batch_size: 1000         # 全量迁移每批条数
  batch_interval_ms: 100   # 每批之间休眠，避免打满 DB
  max_retry: 3             # 补偿重试上限，超过则告警

-- 建在旧库 content_community 中
CREATE TABLE IF NOT EXISTS `migration_failed_log` (
    `id`          BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    `operation`   VARCHAR(16)  NOT NULL COMMENT 'insert / update / delete',
    `table_name`  VARCHAR(64)  NOT NULL COMMENT '目标表，本章固定为 user',
    `record_id`   BIGINT       NOT NULL COMMENT '目标表主键',
    `old_data`    JSON         COMMENT '修改前的数据快照',
    `new_data`    JSON         COMMENT '修改后的数据快照',
    `retry_count` INT NOT NULL DEFAULT 0,
    `last_error`  VARCHAR(512) NOT NULL DEFAULT '',
    `status`      TINYINT NOT NULL DEFAULT 0 COMMENT '0 待重试 1 成功 2 放弃',
    `ctime`       BIGINT NOT NULL DEFAULT 0,
    `utime`       BIGINT NOT NULL DEFAULT 0,
    KEY `idx_status_retry` (`status`, `retry_count`)
) ENGINE=InnoDB;

// internal/migration/failed_log_dao.go
package migration

import (
	"context"
	"time"

	"gorm.io/gorm"
)

// FailedLog 与补偿表 migration_failed_log 一一对应
type FailedLog struct {
	Id         int64  `gorm:"primaryKey"`
	Operation  string // insert / update / delete
	TableName  string `gorm:"column:table_name"` // 目标表，本章固定 user
	RecordId   int64  `gorm:"column:record_id"`  // 目标表主键
	OldData    string // 改前快照 JSON，排查用
	NewData    string // 改后快照 JSON，重试 update 靠它重放
	RetryCount int    `gorm:"column:retry_count"`
	LastError  string `gorm:"column:last_error"`
	Status     int    // 0 待重试 1 成功 2 放弃
	Ctime, Utime int64
}

// TableName 显式指定表名，防止 GORM 按复数规则猜错
func (FailedLog) TableName() string { return "migration_failed_log" }

type FailedLogDAO interface {
	Insert(ctx context.Context, l FailedLog) error                   // 双写失败记一条
	ListPending(ctx context.Context, limit int) ([]FailedLog, error) // 捞待重试记录
	MarkResult(ctx context.Context, id int64, ok bool, msg string) error
}

type GORMFailedLogDAO struct {
	db *gorm.DB // 补偿表在旧库，注入旧库连接
}

func NewGORMFailedLogDAO(db *gorm.DB) FailedLogDAO {
	return &GORMFailedLogDAO{db: db}
}

func (d *GORMFailedLogDAO) Insert(ctx context.Context, l FailedLog) error {
	l.Ctime, l.Utime, l.Status = time.Now().Unix(), time.Now().Unix(), 0
	return d.db.WithContext(ctx).Create(&l).Error
}

func (d *GORMFailedLogDAO) ListPending(ctx context.Context, limit int) ([]FailedLog, error) {
	var res []FailedLog
	// 只捞「没成功且未达重试上限」的记录，按 id 升序 = 先失败先重试
	err := d.db.WithContext(ctx).
		Where("status = 0 AND retry_count < 3").
		Order("id ASC").Limit(limit).Find(&res).Error
	return res, err
}

func (d *GORMFailedLogDAO) MarkResult(ctx context.Context, id int64, ok bool, msg string) error {
	upt := map[string]any{"utime": time.Now().Unix(), "last_error": msg}
	if ok {
		upt["status"] = 1 // 重放成功
	} else {
		upt["retry_count"] = gorm.Expr("retry_count + 1") // 失败次数自增
	}
	return d.db.WithContext(ctx).Model(&FailedLog{}).Where("id = ?", id).Updates(upt).Error
}

// internal/user/repository/double_write_repository.go
package repository

import (
	"context"
	"encoding/json"

	"github.com/yourname/content-community/internal/migration"
	"github.com/yourname/content-community/internal/user/domain"
	"go.uber.org/zap"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

// DoubleWriteUserRepository 嵌入旧库仓储，读透传、写双写
type DoubleWriteUserRepository struct {
	domain.UserRepository             // 旧库仓储（一行不改），读方法全部透传
	dst             *gorm.DB          // 新库连接
	failed          migration.FailedLogDAO // 补偿表 DAO（旧库）
	enable          bool              // migration.double_write 开关
}

func NewDoubleWriteUserRepository(src domain.UserRepository, dst *gorm.DB,
	failed migration.FailedLogDAO, enable bool) domain.UserRepository {
	return &DoubleWriteUserRepository{UserRepository: src, dst: dst, failed: failed, enable: enable}
}

func (r *DoubleWriteUserRepository) Create(ctx context.Context, u *domain.User) (int64, error) {
	id, err := r.UserRepository.Create(ctx, u) // 先写旧库（基准库），它失败才算主流程失败
	if err != nil {
		return 0, err
	}
	if r.enable {
		r.writeDst(ctx, "insert", id, nil, *u)
	}
	return id, nil
}

func (r *DoubleWriteUserRepository) Update(ctx context.Context, u domain.User) error {
	old, err := r.UserRepository.GetByID(ctx, u.Id) // 取旧快照：补偿重试、排查都用
	if err != nil {
		return err // 取不到说明记录不存在
	}
	if err = r.UserRepository.Update(ctx, u); err != nil {
		return err
	}
	if r.enable {
		r.writeDst(ctx, "update", u.Id, &old, u)
	}
	return nil
}

func (r *DoubleWriteUserRepository) Delete(ctx context.Context, id int64) error {
	if err := r.UserRepository.Delete(ctx, id); err != nil {
		return err
	}
	if r.enable {
		r.writeDst(ctx, "delete", id, nil, domain.User{}) // delete 重放只需 record_id
	}
	return nil
}

// writeDst 写新库：失败不打断主流程，只记补偿表 + error 日志（线上再接告警）
func (r *DoubleWriteUserRepository) writeDst(ctx context.Context,
	op string, id int64, old *domain.User, newU domain.User) {
	var err error
	switch op {
	case "insert", "update":
		// UPSERT：有则更新无则插入，天然幂等，补偿重试可安全重放
		err = r.dst.WithContext(ctx).Clauses(clause.OnConflict{UpdateAll: true}).
			Create(&newU).Error
	case "delete":
		err = r.dst.WithContext(ctx).Where("id = ?", id).Delete(&domain.User{}).Error
	}
	if err == nil {
		return
	}
	zap.L().Error("双写新库失败，转入补偿", zap.String("op", op), zap.Int64("id", id), zap.Error(err))
	oldB, _ := json.Marshal(old) // old 为 nil 时序列化成 "null"，不影响排查
	newB, _ := json.Marshal(newU)
	_ = r.failed.Insert(ctx, migration.FailedLog{
		Operation: op, TableName: "user", RecordId: id,
		OldData: string(oldB), NewData: string(newB), LastError: err.Error(),
	})
}

// ReApplyFailed 重放一条补偿记录（定时任务和手动兜底都调它）
func (r *DoubleWriteUserRepository) ReApplyFailed(ctx context.Context, l migration.FailedLog) error {
	if l.Operation == "delete" {
		return r.dst.WithContext(ctx).Where("id = ?", l.RecordId).Delete(&domain.User{}).Error
	}
	var u domain.User
	if err := json.Unmarshal([]byte(l.NewData), &u); err != nil {
		return err
	}
	return r.dst.WithContext(ctx).Clauses(clause.OnConflict{UpdateAll: true}).Create(&u).Error
}

// internal/migration/retry_job.go
package migration

import (
	"context"
	"time"

	"go.uber.org/zap"
)

// ReApplier 双写仓储实现它：能重放一条补偿记录（定义小接口避免反向依赖 user 包）
type ReApplier interface {
	ReApplyFailed(ctx context.Context, l FailedLog) error
}

type RetryJob struct {
	failed FailedLogDAO
	repo   ReApplier
}

func NewRetryJob(failed FailedLogDAO, repo ReApplier) *RetryJob {
	return &RetryJob{failed: failed, repo: repo}
}

func (j *RetryJob) Start(ctx context.Context) {
	go func() {
		ticker := time.NewTicker(5 * time.Minute)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				j.runOnce(ctx)
			}
		}
	}()
}

func (j *RetryJob) runOnce(ctx context.Context) {
	logs, err := j.failed.ListPending(ctx, 100)
	if err != nil {
		zap.L().Error("捞取补偿记录失败", zap.Error(err))
		return
	}
	for _, l := range logs {
		if err = j.repo.ReApplyFailed(ctx, l); err == nil {
			_ = j.failed.MarkResult(ctx, l.Id, true, "")
			continue
		}
		_ = j.failed.MarkResult(ctx, l.Id, false, err.Error())
		if l.RetryCount+1 >= 3 {
			// 达到重试上限仍失败：打告警日志，转人工处理（线上接告警平台）
			zap.L().Error("补偿重试达上限，需人工介入", zap.Int64("id", l.Id))
		}
	}
}

// cmd/migrate/main.go
package main

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/redis/go-redis/v9"
	"github.com/spf13/viper"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
	"github.com/yourname/content-community/internal/user/domain"
)

const checkpointKey = "migration:user:checkpoint" // 断点：已迁移到的最大 id

func main() {
	if len(os.Args) < 2 {
		panic("用法: go run ./cmd/migrate run|verify")
	}
	viper.SetConfigFile("configs/config.yaml")
	if err := viper.ReadInConfig(); err != nil {
		panic("读配置失败: " + err.Error())
	}
	src := mustOpen(viper.GetString("mysql.dsn"))
	dst := mustOpen(viper.GetString("mysql_new.dsn"))
	if os.Args[1] == "run" {
		migrateAll(src, dst)
	} else {
		verify(src, dst)
	}
}

func mustOpen(dsn string) *gorm.DB {
	db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{})
	if err != nil {
		panic("连接数据库失败: " + err.Error())
	}
	return db
}

// migrateAll 全量迁移：游标分页读旧库 → UPSERT 写新库 → Redis 记断点（断点续传）
func migrateAll(src, dst *gorm.DB) {
	ctx := context.Background()
	rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
	lastId, _ := rdb.Get(ctx, checkpointKey).Int64() // 断点续传：从上次位置继续
	for {
		var batch []domain.User
		// 游标分页避免深翻页：只查 id 更大的一批，永远走主键索引
		if err := src.WithContext(ctx).Where("id > ?", lastId).
			Order("id ASC").Limit(1000).Find(&batch).Error; err != nil {
			panic(err)
		}
		if len(batch) == 0 {
			break // 旧库扫完，全量迁移结束
		}
		// UPSERT：存在则全字段更新，重跑、重试都不会主键冲突
		if err := dst.WithContext(ctx).Clauses(clause.OnConflict{UpdateAll: true}).
			Create(&batch).Error; err != nil {
			panic(err) // 本批失败退出；断点在 Redis，重跑续传
		}
		lastId = batch[len(batch)-1].Id
		_ = rdb.Set(ctx, checkpointKey, lastId, 0).Err()
		fmt.Printf("已迁移到 id=%d\n", lastId)
		time.Sleep(100 * time.Millisecond) // 限速，给线上 DB 留喘息空间
	}
	fmt.Println("全量迁移完成")
}

// verify 一致性校验：先比行数，再随机抽 1000 条逐字段比对
func verify(src, dst *gorm.DB) {
	ctx := context.Background()
	var srcCnt, dstCnt int64
	src.WithContext(ctx).Model(&domain.User{}).Count(&srcCnt)
	dst.WithContext(ctx).Model(&domain.User{}).Count(&dstCnt)
	fmt.Printf("行数对比：旧库=%d 新库=%d\n", srcCnt, dstCnt)
	if srcCnt != dstCnt {
		fmt.Println("!! 行数不一致，禁止进入灰度，先排查缺数的主键区间")
		return
	}
	// RAND() 只允许出现在抽样场景，业务代码禁用
	var ids []int64
	src.WithContext(ctx).Model(&domain.User{}).Order("RAND()").Limit(1000).Pluck("id", &ids)
	var a, b []domain.User
	src.WithContext(ctx).Where("id IN ?", ids).Order("id ASC").Find(&a)
	dst.WithContext(ctx).Where("id IN ?", ids).Order("id ASC").Find(&b)
	diff := 0
	// 按 id 建索引再比对：不能按下标直接对位——若新库缺了某条 id，b 比 a 短，
	// 用 a[i]!=b[i] 会下标越界；即便长度相同也可能因 id 错位而误报
	bMap := make(map[int64]domain.User, len(b))
	for _, u := range b {
		bMap[u.Id] = u
	}
	for _, ua := range a {
		ub, ok := bMap[ua.Id]
		if !ok {
			diff++
			fmt.Printf("新库缺失 id=%d：旧=%+v\n", ua.Id, ua)
			continue
		}
		if ua != ub { // 字段都是基础类型，结构体可直接比较
			diff++
			fmt.Printf("不一致 id=%d：旧=%+v 新=%+v\n", ua.Id, ua, ub)
		}
	}
	fmt.Printf("抽样 %d 条，不一致 %d 条\n", len(a), diff)
}

// internal/user/repository/gray_read.go
package repository

import (
	"context"
	"sync/atomic"
	"time"

	"github.com/redis/go-redis/v9"
	"github.com/yourname/content-community/internal/user/domain"
)

const ratioKey = "migration:user:read_ratio"

// GrayReadRouter 嵌入双写仓储：写路径与其余读方法全部透传，只重写 GetByID
type GrayReadRouter struct {
	domain.UserRepository               // 双写仓储（读默认走它透传的旧库）
	dst                 domain.UserRepository // 新库读仓储：同一实现 + 新库连接
	rdb                 *redis.Client
	ratio               atomic.Value // 当前灰度比例 float64，动态刷新
}

func NewGrayReadRouter(dw, dst domain.UserRepository, rdb *redis.Client, def float64) *GrayReadRouter {
	g := &GrayReadRouter{UserRepository: dw, dst: dst, rdb: rdb}
	g.ratio.Store(def) // Redis 没有值时用配置兜底
	go g.loop()
	return g
}

// loop 每 10 秒从 Redis 刷一次比例，调整灰度不用重启服务
func (g *GrayReadRouter) loop() {
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()
	for range ticker.C {
		if v, err := g.rdb.Get(context.Background(), ratioKey).Float64(); err == nil && v >= 0 && v <= 1 {
			g.ratio.Store(v)
		}
	}
}

// GetByID 灰度核心：id 尾号落在灰度桶内 → 读新库。
// 按 user_id 取模保证同一用户永远走同一个库，不会「忽新忽旧」。
func (g *GrayReadRouter) GetByID(ctx context.Context, id int64) (domain.User, error) {
	if float64(id%10000) < g.ratio.Load().(float64)*10000 {
		return g.dst.GetByID(ctx, id)
	}
	return g.UserRepository.GetByID(ctx, id)
}

redis-cli set migration:user:read_ratio 0.01  # 1% 用户读新库，稳定后逐步放大 10% → 50% → 1
redis-cli set migration:user:read_ratio 0     # 发现异常：一键回滚，全部读旧库

// internal/main.go（组装层示意）
func main() {
    // ... 略去 viper / redis / gorm 初始化 ...
    srcRepo := repository.NewGORMUserRepository(srcDB)   // 旧库仓储
    dstRepo := repository.NewGORMUserRepository(dstDB)   // 新库仓储（同一实现换连接）
    failedDAO := migration.NewGORMFailedLogDAO(srcDB)    // 补偿表 DAO（建在旧库）

    // 双写仓储：包一层，写走双写、读透传旧库
    dw := repository.NewDoubleWriteUserRepository(srcRepo, dstDB, failedDAO, cfg.Migration.DoubleWrite)
    // 补偿定时任务：每 5 分钟重放失败记录
    migration.NewRetryJob(failedDAO, dw).Start(ctx)
    // 灰度读路由：写走双写、读按 user_id 取模切新库，比例存 Redis 动态刷新
    gray := repository.NewGrayReadRouter(dw, dstRepo, rdb, cfg.Migration.ReadNewRatio)

    // 把 gray 当 domain.UserRepository 注入 Service：上层业务代码一行不用改
    userSvc := service.NewUserService(gray)
}

content-community
├── cmd/migrate/main.go                  # 【新建】全量迁移 + 一致性校验命令
├── configs/config.yaml                  # 【修改】新增 mysql_new、migration 配置段
├── schema/user_service.sql              # 【新建】新库建表脚本
└── internal
    ├── migration/                       # 【新建】迁移公共件
    │   ├── failed_log_dao.go            #    补偿表 DAO
    │   └── retry_job.go                 #    补偿重试定时任务
    └── user
        └── repository
            ├── double_write_repository.go  # 【新建】双写仓储
            └── gray_read.go                # 【新建】灰度读路由