package impl

import (
	"context"
	"fmt"
	"strconv"
	"strings"

	"github.com/AlekSi/pointer"
	"github.com/elliotchance/pie/v2"
	"github.com/jinzhu/copier"
	workflowEntity "github.com/shiningrush/fastflow/pkg/entity"
	appconfig "github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/internal/status_matrix"
	"github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	crrfRepo "github.com/tidbcloud/dataflow-service/app/repo/crrf"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	service_config "github.com/tidbcloud/dataflow-service/internal/config"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
)

type regionModifierType string

const (
	typeAWSEnableDualRegion  regionModifierType = "awsEnableDualRegion"
	typeAWSDisableDualRegion regionModifierType = "awsDisableDualRegion"
	typeGCPToggleDualRegion  regionModifierType = "gcpToggleDualRegion"

	enableDualRegionSupportLen = 2
)

type regionModifierProxy struct {
	awsEnableDualRegion  regionModifier
	awsDisableDualRegion regionModifier
	gcpToggleDualRegion  regionModifier
}

func newRegionModifierProxy(
	dbProvider db.TransProvider,
	backupSettingRepo backupRepo.BackupSettingRepo,
	logBackupRepo backupRepo.LogBackupRepo,
	workflowClient workflow.Workflow,
	statusMatrix status_matrix.StatusMatrix,
	tidbClusterInfraClient grpcli.TiDBClusterInfraClient,
	pitBackupSwitcher backup.PitBackupSwitcher,
	crrfSettingRepo crrfRepo.CRRFSettingRepo,
) *regionModifierProxy {
	return &regionModifierProxy{
		awsEnableDualRegion: newAWSEnableDualRegion(
			dbProvider,
			backupSettingRepo,
			workflowClient,
			statusMatrix,
			tidbClusterInfraClient,
			pitBackupSwitcher,
			logBackupRepo,
		),
		awsDisableDualRegion: newAWSDisableDualRegion(
			dbProvider,
			backupSettingRepo,
			workflowClient,
			statusMatrix,
			tidbClusterInfraClient,
			pitBackupSwitcher,
			logBackupRepo,
			crrfSettingRepo,
		),
		gcpToggleDualRegion: newGCPRegionModifierImpl(
			dbProvider,
			backupSettingRepo,
			workflowClient,
			statusMatrix,
			tidbClusterInfraClient,
			pitBackupSwitcher,
			logBackupRepo,
			crrfSettingRepo,
		),
	}
}

func (p *regionModifierProxy) get(
	req *domain.RegionSettingModificationReqV2, infraProviderType domain.InfraProviderType,
) regionModifier {
	// for clusters on GCP
	if infraProviderType == domain.InfraProviderTypeGCP || infraProviderType == domain.InfraProviderTypeGKE {
		return p.gcpToggleDualRegion
	}

	// for clusters on AWS
	if req.Enable {
		return p.awsEnableDualRegion
	}
	return p.awsDisableDualRegion
}

type regionModifier interface {
	execute(
		ctx context.Context, req *domain.RegionSettingModificationReqV2,
	) (*domain.RegionSettingModificationRespV2, bizerr.BizError)
	businessType() regionModifierType
}

type regionModifierHook interface {
	businessType() regionModifierType
	paramCheck(
		ctx context.Context, req *domain.RegionSettingModificationReqV2, setting *entity.BackupSetting) bizerr.BizError
	updateEntity(
		ctx context.Context, req *domain.RegionSettingModificationReqV2, setting *entity.BackupSetting,
	) (*entity.BackupSetting, bizerr.BizError)
	buildDagVars(ctx context.Context, req *buildDagVarsReq) (string, map[string]string, bizerr.BizError)
}

type buildDagVarsReq struct {
	oldBackupSetting *entity.BackupSetting
	newBackupSetting *entity.BackupSetting
}

type regionModifierImpl struct {
	dbProvider        db.TransProvider
	backupSettingRepo backupRepo.BackupSettingRepo
	logBackupRepo     backupRepo.LogBackupRepo

	workflowClient workflow.Workflow

	statusMatrix      status_matrix.StatusMatrix
	pitBackupSwitcher backup.PitBackupSwitcher

	hook regionModifierHook
}

func newAWSEnableDualRegion(
	dbProvider db.TransProvider,
	backupSettingRepo backupRepo.BackupSettingRepo,
	workflowClient workflow.Workflow,
	statusMatrix status_matrix.StatusMatrix,
	tidbClusterInfraClient grpcli.TiDBClusterInfraClient,
	pitBackupSwitcher backup.PitBackupSwitcher,
	logBackupRepo backupRepo.LogBackupRepo,
) regionModifier {
	hook := newAWSEnableDualRegionHook(backupSettingRepo, tidbClusterInfraClient, statusMatrix)
	return newRegionModifierImpl(dbProvider, backupSettingRepo, workflowClient, statusMatrix, pitBackupSwitcher,
		logBackupRepo, hook)
}

func newAWSDisableDualRegion(
	dbProvider db.TransProvider,
	backupSettingRepo backupRepo.BackupSettingRepo,
	workflowClient workflow.Workflow,
	statusMatrix status_matrix.StatusMatrix,
	tidbClusterInfraClient grpcli.TiDBClusterInfraClient,
	pitBackupSwitcher backup.PitBackupSwitcher,
	logBackupRepo backupRepo.LogBackupRepo,
	crrfSettingRepo crrfRepo.CRRFSettingRepo,
) regionModifier {
	hook := newAWSDisableDualRegionHook(backupSettingRepo, tidbClusterInfraClient, statusMatrix, crrfSettingRepo)
	return newRegionModifierImpl(dbProvider, backupSettingRepo, workflowClient, statusMatrix, pitBackupSwitcher,
		logBackupRepo, hook)
}

func newGCPRegionModifierImpl(
	dbProvider db.TransProvider,
	backupSettingRepo backupRepo.BackupSettingRepo,
	workflowClient workflow.Workflow,
	statusMatrix status_matrix.StatusMatrix,
	tidbClusterInfraClient grpcli.TiDBClusterInfraClient,
	pitBackupSwitcher backup.PitBackupSwitcher,
	logBackupRepo backupRepo.LogBackupRepo,
	crrfSettingRepo crrfRepo.CRRFSettingRepo,
) regionModifier {
	hook := newGCPRegionModifierHook(backupSettingRepo, tidbClusterInfraClient, statusMatrix, crrfSettingRepo)
	return newRegionModifierImpl(dbProvider, backupSettingRepo, workflowClient, statusMatrix, pitBackupSwitcher,
		logBackupRepo, hook)
}

func newRegionModifierImpl(
	dbProvider db.TransProvider,
	backupSettingRepo backupRepo.BackupSettingRepo,
	workflowClient workflow.Workflow,
	statusMatrix status_matrix.StatusMatrix,
	pitBackupSwitcher backup.PitBackupSwitcher,
	logBackupRepo backupRepo.LogBackupRepo,
	hook regionModifierHook,
) regionModifier {
	return &regionModifierImpl{
		dbProvider:        dbProvider,
		backupSettingRepo: backupSettingRepo,
		logBackupRepo:     logBackupRepo,
		workflowClient:    workflowClient,
		statusMatrix:      statusMatrix,
		pitBackupSwitcher: pitBackupSwitcher,
		hook:              hook,
	}
}

func (m *regionModifierImpl) businessType() regionModifierType {
	return m.hook.businessType()
}

func (m *regionModifierImpl) execute(
	ctx context.Context, req *domain.RegionSettingModificationReqV2,
) (*domain.RegionSettingModificationRespV2, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	backupSetting, err := m.queryBackupSetting(ctx, req.ClusterID)
	if err != nil {
		return nil, err
	}
	err = m.hook.paramCheck(ctx, req, backupSetting)
	if err != nil {
		return nil, err
	}
	err = m.isActionFriendly2Status(ctx, backupSetting)
	if err != nil {
		return nil, err
	}

	if req.DryRun {
		logger.Info("dry run success")
		return nil, nil
	}

	var dagIns *workflowEntity.DagInstance
	dbErr := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		oldBackupSetting := new(entity.BackupSetting)
		err := copier.Copy(oldBackupSetting, backupSetting)
		if err != nil {
			logger.Warn("copy backup setting failed", zap.Error(err))
			return err
		}
		newBackupSetting, err := m.hook.updateEntity(ctx, req, backupSetting)
		if err != nil {
			return err
		}
		if err := m.restartOrCancelRestartPitBackupWhenNeed(ctx, newBackupSetting); err != nil {
			return err
		}
		dagIns, err = m.createDagIns(ctx, req, oldBackupSetting, newBackupSetting)
		if err != nil {
			return err
		}
		return nil
	})
	if dbErr != nil {
		logger.Error("modify bucket setting failed", zap.Any("request", req), zap.Error(dbErr))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return &domain.RegionSettingModificationRespV2{
		WorkflowID: pointer.ToOrNil(dagIns.ID),
	}, nil
}

func (m *regionModifierImpl) createDagIns(
	ctx context.Context, req *domain.RegionSettingModificationReqV2, oldBackupSetting *entity.BackupSetting,
	newBackupSetting *entity.BackupSetting,
) (*workflowEntity.DagInstance, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	dagName, dagInsVars, err := m.hook.buildDagVars(ctx, &buildDagVarsReq{
		oldBackupSetting: oldBackupSetting,
		newBackupSetting: newBackupSetting,
	})
	if err != nil {
		logger.Error("build dag vars failed", zap.Any("backup setting id ", newBackupSetting.ID), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	dagIns, dagErr := m.workflowClient.RunDagWithTags(
		dagName, dagInsVars,
		map[string]string{
			workflow.BusinessTypeKey:   string(workflow.BusinessTypeBackupSetting),
			workflow.BusinessActionKey: string(workflow.BusinessTypeBackupSettingModifyBucketRegion),
			workflow.BusinessID:        strconv.FormatUint(uint64(newBackupSetting.ID), 10),
			workflow.ClusterID:         strconv.FormatUint(uint64(newBackupSetting.ClusterID), 10),
		})
	if dagErr != nil {
		logger.Error("enable dual region dag failed",
			zap.Any("backup setting id", newBackupSetting.ID), zap.Error(dagErr))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	logger.Info("enable dual region dag started",
		zap.Any("request", req), zap.Any("dag id", dagIns.ID))
	return dagIns, nil
}

func (m *regionModifierImpl) queryBackupSetting(
	ctx context.Context, clusterID *uint64,
) (*entity.BackupSetting, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	backupSetting, err := m.backupSettingRepo.Get(ctx,
		func(qs entity.BackupSettingQuerySet) entity.BackupSettingQuerySet {
			return qs.ClusterIDEq(entity.ID(*clusterID))
		})
	if err != nil {
		logger.Error("load backup setting failed", zap.Any("cluster id", clusterID), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return backupSetting, nil
}

func (m *regionModifierImpl) isActionFriendly2Status(
	ctx context.Context, backupSetting *entity.BackupSetting,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	isFriendly := m.statusMatrix.IsActionFriendly2Status(status_matrix.CurrentStatus{
		Status:  string(backupSetting.Status),
		Actions: backupSetting.Action.CurrentActions,
	}, string(entity.BackupSettingActionModifyRegion))
	if !isFriendly {
		logger.Warn("modify backup setting failed, status is not friendly",
			zap.Any("backup setting", backupSetting))
		return bizerr_wrapper.NewInvalidInputError(
			appconfig.ErrActionParam, backupSetting.Status, entity.BackupSettingActionModifyRegion)
	}
	return nil
}

func (m *regionModifierImpl) restartOrCancelRestartPitBackupWhenNeed(
	ctx context.Context, newBackupSetting *entity.BackupSetting,
) bizerr.BizError {
	if !newBackupSetting.IsPiTREnabled() {
		return nil
	}
	switchReq := domain.SwitchPitBackupReq{
		ClusterID: newBackupSetting.ClusterID.Uint64(),
	}
	pitBackupSwitchResp, err := m.pitBackupSwitcher.GetOrCreatePitBackupSwitch(ctx, &switchReq)
	if err != nil {
		return err
	}
	pitBackupSwitch := pitBackupSwitchResp.PitBackupSwitch

	// already in restart status,.if we switch back, we need to cancel the restart op
	if pitBackupSwitch.Status == entity.PitBackupStatusWaitingForRestarting {
		logBackup, err := m.getActiveLogBackup(ctx, newBackupSetting)
		if err != nil {
			return err
		}
		if m.checkLogBackupRegionIsSameToSetting(ctx, newBackupSetting, logBackup) {
			log.GetLogger(ctx).Info("log backup region is same to setting, cancel restart pit backup")
			cancelReq := &domain.CancelRestartPitBackupReq{SwitchPitBackupReq: switchReq}
			if err := m.pitBackupSwitcher.CancelRestart(ctx, cancelReq); err != nil {
				return err
			}
		}
		return nil
	}

	// mark the pit backup switch to restart,and when next snapshot backup, it will restart
	restartReq := &domain.RestartPitBackupReq{SwitchPitBackupReq: switchReq}
	if _, err := m.pitBackupSwitcher.Restart(ctx, restartReq); err != nil {
		return err
	}
	return nil
}

func (m *regionModifierImpl) checkLogBackupRegionIsSameToSetting(
	ctx context.Context, newBackupSetting *entity.BackupSetting, logBackup *entity.LogBackup,
) bool {
	if logBackup == nil || logBackup.BucketSettingSnapshot == nil {
		return false
	}
	var logBackupRegions []string
	pie.Of(logBackup.BucketSettingSnapshot.Buckets).Each(func(bucket entity.BackupSettingBucket) {
		logBackupRegions = append(logBackupRegions, bucket.Regions...)
	})
	var backupSettingRegions []string
	pie.Of(newBackupSetting.Bucket.Buckets).Each(func(bucket entity.BackupSettingBucket) {
		backupSettingRegions = append(backupSettingRegions, bucket.Regions...)
	})
	return utils.AreSlicesEqual(logBackupRegions, backupSettingRegions)
}

func (m *regionModifierImpl) getActiveLogBackup(
	ctx context.Context, backupSetting *entity.BackupSetting,
) (*entity.LogBackup, bizerr.BizError) {
	filterF := func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
		qs = qs.ClusterIDEq(backupSetting.ClusterID).
			ProjectIDEq(backupSetting.ProjectID).
			TenantIDEq(backupSetting.TenantID).
			DeletedAtIsNull().
			StatusNotIn(entity.LogBackupStatusStopped).
			PreloadBackupSchedule()
		return qs
	}
	logBackups, err := m.logBackupRepo.List(ctx, filterF)
	if err != nil {
		log.GetLogger(ctx).Error("list log backup error", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrMsgInternalError, backupSetting.ClusterID)
	}
	if len(logBackups) > 1 {
		log.GetLogger(ctx).Error("more than one active log backup existed this should not happen", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrMsgInternalError, backupSetting.ClusterID)
	}
	if len(logBackups) == 0 {
		log.GetLogger(ctx).Warn("no active log backup existed this should not happen", zap.Error(err))
		return nil, nil
	}
	return &logBackups[0], nil
}

type awsEnableDualRegionHook struct {
	baseRegionModifierHook
}

func newAWSEnableDualRegionHook(
	backupSettingRepo backupRepo.BackupSettingRepo,
	tidbClusterInfraClient grpcli.TiDBClusterInfraClient,
	statusMatrix status_matrix.StatusMatrix,
) regionModifierHook {
	return &awsEnableDualRegionHook{
		baseRegionModifierHook: baseRegionModifierHook{
			backupSettingRepo:      backupSettingRepo,
			tidbClusterInfraClient: tidbClusterInfraClient,
			statusMatrix:           statusMatrix,
		},
	}
}

func (e *awsEnableDualRegionHook) businessType() regionModifierType {
	return typeAWSEnableDualRegion
}

func (e *awsEnableDualRegionHook) paramCheck(
	ctx context.Context, req *domain.RegionSettingModificationReqV2, setting *entity.BackupSetting,
) bizerr.BizError {
	return e.enableDualRegionParamCheck(ctx, req, setting)
}

func (e *awsEnableDualRegionHook) updateEntity(
	ctx context.Context, req *domain.RegionSettingModificationReqV2, setting *entity.BackupSetting,
) (*entity.BackupSetting, bizerr.BizError) {
	bucketSetting := &entity.BucketSetting{
		DualRegionEnable: true,
	}
	for i, region := range req.Regions {
		bucketSetting.Buckets = append(bucketSetting.Buckets, entity.BackupSettingBucket{
			Regions: []string{region},
		})
		if i == 0 {
			bucketSetting.Buckets[i].Master = true
		}
	}
	return e.updateBucketSetting(ctx, setting, bucketSetting)
}

func (e *awsEnableDualRegionHook) buildDagVars(
	ctx context.Context, req *buildDagVarsReq,
) (string, map[string]string, bizerr.BizError) {
	newBackupSetting := req.newBackupSetting
	dagInsVars := map[string]string{}
	dagInsVars[workflow.KeyBackupSettingId] = fmt.Sprintf("%d", newBackupSetting.ID)
	masterBucket, err := e.queryMasterBucket(ctx, newBackupSetting)
	if err != nil {
		return "", nil, err
	}
	dagInsVars[workflow.KeyRegion] = masterBucket.Regions[0]
	secondaryBucket, err := e.querySecondaryBucket(ctx, newBackupSetting)
	if err != nil {
		return "", nil, err
	}
	dagInsVars[workflow.KeySecondaryRegion] = secondaryBucket.Regions[0]

	networkReq := &grpcli.RegionalNetworkReq{
		ProjectID:    newBackupSetting.ProjectID.Uint64(),
		ProviderType: newBackupSetting.InfraProviderType,
		Region:       masterBucket.Regions[0],
	}
	nodeRole, infraErr := e.tidbClusterInfraClient.GetNetworkNodeRole(ctx, networkReq)
	if infraErr != nil {
		log.GetLogger(ctx).Warn("get node role error", zap.Error(infraErr))
		return "", nil, bizerr.WrapError(infraErr, "get node role error")
	}
	dagInsVars[workflow.KeyNetworkNodeRole] = nodeRole
	return workflow.DagEnableDualRegionBackupAWS, dagInsVars, nil
}

type awsDisableDualRegionHook struct {
	baseRegionModifierHook
}

func newAWSDisableDualRegionHook(
	backupSettingRepo backupRepo.BackupSettingRepo,
	tidbClusterInfraClient grpcli.TiDBClusterInfraClient,
	statusMatrix status_matrix.StatusMatrix,
	crrfSettingRepo crrfRepo.CRRFSettingRepo,
) regionModifierHook {
	return &awsDisableDualRegionHook{
		baseRegionModifierHook: baseRegionModifierHook{
			backupSettingRepo:      backupSettingRepo,
			tidbClusterInfraClient: tidbClusterInfraClient,
			statusMatrix:           statusMatrix,
			crrfSettingRepo:        crrfSettingRepo,
		},
	}
}

func (e *awsDisableDualRegionHook) businessType() regionModifierType {
	return typeAWSDisableDualRegion
}

func (e *awsDisableDualRegionHook) paramCheck(
	ctx context.Context, req *domain.RegionSettingModificationReqV2, setting *entity.BackupSetting,
) bizerr.BizError {
	return e.disableDualRegionParamCheck(ctx, req, setting)
}

func (e *awsDisableDualRegionHook) updateEntity(
	ctx context.Context, req *domain.RegionSettingModificationReqV2, setting *entity.BackupSetting,
) (*entity.BackupSetting, bizerr.BizError) {
	bucketSetting := &entity.BucketSetting{
		DualRegionEnable: false,
		Buckets: []entity.BackupSettingBucket{
			{
				Regions: []string{req.Regions[0]},
				Master:  true,
			},
		},
	}
	return e.updateBucketSetting(ctx, setting, bucketSetting)
}

func (e *awsDisableDualRegionHook) buildDagVars(
	ctx context.Context, req *buildDagVarsReq,
) (string, map[string]string, bizerr.BizError) {
	dagInsVars := map[string]string{}
	dagInsVars[workflow.KeyBackupSettingId] = fmt.Sprintf("%d", req.newBackupSetting.ID)
	newMasterBucket, err := e.queryMasterBucket(ctx, req.newBackupSetting)
	if err != nil {
		return "", nil, err
	}
	dagInsVars[workflow.KeyRegion] = newMasterBucket.Regions[0]

	networkReq := &grpcli.RegionalNetworkReq{
		ProjectID:    req.newBackupSetting.ProjectID.Uint64(),
		ProviderType: req.newBackupSetting.InfraProviderType,
		Region:       newMasterBucket.Regions[0],
	}
	nodeRole, infraErr := e.tidbClusterInfraClient.GetNetworkNodeRole(ctx, networkReq)
	if infraErr != nil {
		log.GetLogger(ctx).Warn("get node role error", zap.Error(infraErr))
		return "", nil, bizerr.WrapError(infraErr, "get node role error")
	}
	dagInsVars[workflow.KeyNetworkNodeRole] = nodeRole

	oldMasterBucket, err := e.queryMasterBucket(ctx, req.oldBackupSetting)
	if err != nil {
		return "", nil, err
	}
	dagInsVars[workflow.KeyMasterBucketId] = strconv.FormatUint(oldMasterBucket.BucketId, appconfig.DecimalBase)
	oldSecondaryBucket, err := e.querySecondaryBucket(ctx, req.oldBackupSetting)
	if err != nil {
		return "", nil, err
	}
	dagInsVars[workflow.KeySecondaryBucketId] = strconv.FormatUint(oldSecondaryBucket.BucketId, appconfig.DecimalBase)
	if req.oldBackupSetting.Bucket.Prefix != nil {
		dagInsVars[workflow.KeyBucketPrefix] = *req.oldBackupSetting.Bucket.Prefix
	}
	return workflow.DagDisableDualRegionBackupAWS, dagInsVars, nil
}

type gcpRegionModifierHook struct {
	baseRegionModifierHook
}

func newGCPRegionModifierHook(
	backupSettingRepo backupRepo.BackupSettingRepo,
	tidbClusterInfraClient grpcli.TiDBClusterInfraClient,
	statusMatrix status_matrix.StatusMatrix,
	crrfSettingRepo crrfRepo.CRRFSettingRepo,
) regionModifierHook {
	return &gcpRegionModifierHook{
		baseRegionModifierHook: baseRegionModifierHook{
			backupSettingRepo:      backupSettingRepo,
			tidbClusterInfraClient: tidbClusterInfraClient,
			statusMatrix:           statusMatrix,
			crrfSettingRepo:        crrfSettingRepo,
		},
	}
}

func (m *gcpRegionModifierHook) businessType() regionModifierType {
	return typeGCPToggleDualRegion
}

func (m *gcpRegionModifierHook) paramCheck(
	ctx context.Context, req *domain.RegionSettingModificationReqV2, setting *entity.BackupSetting,
) bizerr.BizError {
	if req.Enable {
		if err := m.enableDualRegionParamCheck(ctx, req, setting); err != nil {
			return err
		}
		if !m.isDualRegionLegal(req.Regions[0], req.Regions[1]) {
			log.GetLogger(ctx).Warn("modify backup setting failed, regions are illegal in option config",
				zap.Any("regions", req.Regions))
			return bizerr_wrapper.NewInvalidInputError(appconfig.ErrInvalidParam, "regions", req.Regions)
		}
		return nil
	} else {
		return m.disableDualRegionParamCheck(ctx, req, setting)
	}
}

func (m *gcpRegionModifierHook) isDualRegionLegal(region1, region2 string) bool {
	bucketOption := service_config.Global().ServerConfig.BucketOptions
	dualRegionCandidateSet := bucketOption.GetGCPDualRegionCandidateSet(region1)
	_, ok := dualRegionCandidateSet[region2]
	return ok
}

func (m *gcpRegionModifierHook) updateEntity(
	ctx context.Context, req *domain.RegionSettingModificationReqV2, setting *entity.BackupSetting,
) (*entity.BackupSetting, bizerr.BizError) {
	bucketSetting := &entity.BucketSetting{
		DualRegionEnable: req.Enable,
		Buckets: []entity.BackupSettingBucket{
			{
				Regions: req.Regions,
				Master:  true,
			},
		},
	}
	return m.updateBucketSetting(ctx, setting, bucketSetting)
}

func (m *gcpRegionModifierHook) buildDagVars(
	ctx context.Context, req *buildDagVarsReq,
) (string, map[string]string, bizerr.BizError) {
	newBackupSetting := req.newBackupSetting
	if len(newBackupSetting.Bucket.Buckets) == 0 {
		return "", nil, bizerr_wrapper.NewInternalErr("master bucket not found")
	}

	newBucket := newBackupSetting.Bucket.Buckets[0]
	dagInsVars := map[string]string{}
	dagInsVars[workflow.KeyBackupSettingId] = fmt.Sprintf("%d", newBackupSetting.ID)
	dagInsVars[workflow.KeyRegion] = strings.Join(newBucket.Regions, workflow.SepBucketRegions)

	networkReq := &grpcli.RegionalNetworkReq{
		ProjectID:    req.newBackupSetting.ProjectID.Uint64(),
		ProviderType: req.newBackupSetting.InfraProviderType,
		Region:       newBucket.Regions[0],
	}
	nodeRole, infraErr := m.tidbClusterInfraClient.GetNetworkNodeRole(ctx, networkReq)
	if infraErr != nil {
		log.GetLogger(ctx).Warn("get node role error", zap.Error(infraErr))
		return "", nil, bizerr.WrapError(infraErr, "get node role error")
	}
	dagInsVars[workflow.KeyNetworkNodeRole] = nodeRole
	return workflow.DagToggleDualRegionBackupGCP, dagInsVars, nil
}

type baseRegionModifierHook struct {
	backupSettingRepo      backupRepo.BackupSettingRepo
	tidbClusterInfraClient grpcli.TiDBClusterInfraClient
	statusMatrix           status_matrix.StatusMatrix
	crrfSettingRepo        crrfRepo.CRRFSettingRepo
}

func (m *baseRegionModifierHook) enableDualRegionParamCheck(
	ctx context.Context, req *domain.RegionSettingModificationReqV2, setting *entity.BackupSetting,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	if setting.Bucket.DualRegionEnable {
		logger.Warn("modify backup setting failed, dual region already enabled")
		return bizerr_wrapper.NewInvalidInputError(appconfig.ErrInvalidParam, "dual_region_enable", true)
	}

	if len(req.Regions) != enableDualRegionSupportLen {
		logger.Warn("modify backup setting failed, regions must be 2 when enable")
		return bizerr_wrapper.NewInvalidInputError(appconfig.ErrInvalidParam, "regions", req.Regions)
	}

	if req.Regions[0] == req.Regions[1] {
		logger.Warn("modify backup setting failed, regions must be different")
		return bizerr_wrapper.NewInvalidInputError(appconfig.ErrInvalidParam, "regions", req.Regions)
	}

	masterBucket, err := m.queryMasterBucket(ctx, setting)
	if err != nil {
		return err
	}

	if req.Regions[0] != masterBucket.Regions[0] {
		logger.Warn("modify backup setting failed, regions must contain master region",
			zap.Any("regions", req.Regions))
		return bizerr_wrapper.NewInvalidInputError(appconfig.ErrInvalidParam, "regions", req.Regions)
	}
	return nil
}

func (m *baseRegionModifierHook) disableDualRegionParamCheck(
	ctx context.Context, req *domain.RegionSettingModificationReqV2, setting *entity.BackupSetting,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	if !setting.Bucket.DualRegionEnable {
		logger.Warn("modify backup setting failed, dual region already disabled")
		return bizerr_wrapper.NewInvalidInputError(appconfig.ErrInvalidParam, "dual_region_enable", false)
	}

	if len(req.Regions) != 1 {
		logger.Warn("modify backup setting failed, regions must be 1 when disable")
		return bizerr_wrapper.NewInvalidInputError(appconfig.ErrInvalidParam, "regions", req.Regions)
	}

	masterBucket, err := m.queryMasterBucket(ctx, setting)
	if err != nil {
		return err
	}

	if req.Regions[0] != masterBucket.Regions[0] {
		logger.Warn("modify backup setting failed, regions must contain master region",
			zap.Any("regions", req.Regions))
		return bizerr_wrapper.NewInvalidInputError(appconfig.ErrInvalidParam, "regions", req.Regions)
	}
	return nil
}

func (m *baseRegionModifierHook) updateBucketSetting(
	ctx context.Context, oldSetting *entity.BackupSetting, newBucketSetting *entity.BucketSetting,
) (*entity.BackupSetting, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	nextStatus := m.statusMatrix.CalculateNextStatusForAction(status_matrix.CurrentStatus{
		Status:  string(oldSetting.Status),
		Actions: oldSetting.Action.CurrentActions,
	}, string(entity.BackupSettingActionModifyRegion))
	oldSetting.Status = entity.BackupSettingStatus(nextStatus.Status)
	oldSetting.Action.CurrentActions = nextStatus.Actions
	oldSetting.Bucket = newBucketSetting
	err := oldSetting.RefreshBucketPrefix()
	if err != nil {
		logger.Error("refresh bucket prefix failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	logger.Info("update backup setting", zap.Any("bucket setting", oldSetting))
	err = m.backupSettingRepo.Save(ctx, oldSetting)
	if err != nil {
		logger.Error("update backup setting failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return oldSetting, nil
}

func (m *baseRegionModifierHook) queryMasterBucket(ctx context.Context,
	setting *entity.BackupSetting,
) (*entity.BackupSettingBucket, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	masterBucket := setting.QueryMasterBucket()
	if masterBucket == nil {
		logger.Error("load backup setting failed, master bucket not found",
			zap.Any("backup setting", setting))
		return nil, bizerr_wrapper.NewInternalErr("master bucket not found")
	}
	return masterBucket, nil
}

func (m *baseRegionModifierHook) querySecondaryBucket(ctx context.Context,
	setting *entity.BackupSetting,
) (*entity.BackupSettingBucket, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	secondaryBucket := setting.QuerySecondaryBucket()
	if secondaryBucket == nil {
		logger.Error("load backup setting failed, secondary bucket not found",
			zap.Any("backup setting", setting))
		return nil, bizerr_wrapper.NewInternalErr("secondary bucket not found")
	}
	return secondaryBucket, nil
}