package impl

import (
	"context"
	"errors"
	"strconv"

	"github.com/pelletier/go-toml/v2"
	"github.com/tidbcloud/dataflow-service/app/internal/feature_gate"
	"github.com/tidbcloud/dataflow-service/app/models/config_manager"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc_spec"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/structpb"
)

const (
	diskThroughput400MBs  = 400
	diskThroughput500MBs  = 500
	diskThroughput1000MBs = 1000

	diskIOPS4000  = 4000
	diskIOPS6000  = 6000
	diskIOPS12000 = 12000
)

var newKernelCPUCore2DiskIOMap = map[uint32]struct {
	ThroughputMbs uint32
	IOPS          uint32
}{
	2:  {ThroughputMbs: diskThroughput400MBs, IOPS: diskIOPS4000},
	4:  {ThroughputMbs: diskThroughput400MBs, IOPS: diskIOPS4000},
	8:  {ThroughputMbs: diskThroughput500MBs, IOPS: diskIOPS6000},
	16: {ThroughputMbs: diskThroughput1000MBs, IOPS: diskIOPS12000},
	32: {ThroughputMbs: diskThroughput1000MBs, IOPS: diskIOPS12000},
	64: {ThroughputMbs: diskThroughput1000MBs, IOPS: diskIOPS12000},
}

type clusterModelRegionalImpl struct {
	instanceTypeModel      ticdc_spec.CDCInstanceTypeModel
	ticdcClusterCli        regionalapi.TiCDCClusterClient
	transProvider          db.TransProvider
	ticdcClusterRepo       repo.TiCDCClusterRepo
	ticdcCustomStorageRepo repo.TiCDCCustomStorageRepo
	configManager          config_manager.TiCDCClusterConfigManager
	regionCli              grpcli.RegionGrpcClient
	featureGate            feature_gate.FeatureGate
}

func NewTiCDCClusterRegionalModel(
	transProvider db.TransProvider,
	instanceTypeModel ticdc_spec.CDCInstanceTypeModel,
	ticdcClusterCli regionalapi.TiCDCClusterClient,
	ticdcClusterRepo repo.TiCDCClusterRepo,
	ticdcCustomStorageRepo repo.TiCDCCustomStorageRepo,
	configManager config_manager.TiCDCClusterConfigManager,
	regionCli grpcli.RegionGrpcClient,
	featureGate feature_gate.FeatureGate,
) ticdc.ClusterModel {
	return &clusterModelRegionalImpl{
		instanceTypeModel:      instanceTypeModel,
		ticdcClusterCli:        ticdcClusterCli,
		transProvider:          transProvider,
		ticdcClusterRepo:       ticdcClusterRepo,
		ticdcCustomStorageRepo: ticdcCustomStorageRepo,
		configManager:          configManager,
		regionCli:              regionCli,
		featureGate:            featureGate,
	}
}

func (m *clusterModelRegionalImpl) Create(
	ctx context.Context, req *domain.CreateTiCDCClusterReq,
) (*domain.CreateTiCDCClusterResp, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	ticdcClusterEnt, err := m.ticdcClusterRepo.Get(
		ctx, func(qs entity.TiCDCClusterQuerySet) entity.TiCDCClusterQuerySet {
			return qs.NameEq(req.Name).ClusterIDEq(entity.ID(req.ClusterID)).
				DeployTypeEq(req.DeployType).DeletedAtIsNull()
		})
	if err != nil && err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return nil, err.Chain("get ticdc cluster entity")
	}
	if ticdcClusterEnt != nil {
		if ticdcClusterEnt.Status == entity.TiCDCClusterStatusDeleting {
			return nil, bizerr.New(bizerr.ErrCodeUnknown,
				"ticdc cluster %d is deleting, wait for it deleted", ticdcClusterEnt.ID.Uint64())
		}
		logger.Info("ticdc cluster already exists", zap.Uint64("cluster_id", ticdcClusterEnt.ID.Uint64()))
		return &domain.CreateTiCDCClusterResp{
			ID: ticdcClusterEnt.ID.Uint64(),
		}, nil
	}

	logicalRegion, err := m.regionCli.GetRegionByTenantID(ctx, req.TenantID, "", req.Region)
	if err != nil {
		return nil, err.Chain("get logical region")
	}
	ticdcClusterEnt = m.createReq2Entity(req, logicalRegion.GetRegion().GetLogicalName())
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.doCreate(ctx, req, ticdcClusterEnt)
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	return &domain.CreateTiCDCClusterResp{
		ID: ticdcClusterEnt.ID.Uint64(),
	}, nil
}

func (m *clusterModelRegionalImpl) doCreate(ctx context.Context,
	req *domain.CreateTiCDCClusterReq,
	ticdcClusterEnt *entity.TiCDCCluster,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	tidbVersion, verr := utils.NewVersion(ticdcClusterEnt.Version)
	if verr != nil {
		logger.Error("tidb version is invalid", zap.String("tidb_version", ticdcClusterEnt.Version), zap.Error(verr))
		return bizerr.WrapErrorWithCode(verr, bizerr.ErrCodeParams, "tidb version is invalid: %s", ticdcClusterEnt.Version)
	}

	isNewKernelEnabled := false
	// New kernel is enabled by default after the compatible TiDB version gate,
	// unless the cluster matches the blacklist. Migration can still force it on.
	if m.isNewKernelEnabled(
		ticdcClusterEnt.TenantID.Uint64(), ticdcClusterEnt.ProjectID.Uint64(), ticdcClusterEnt.ClusterID.Uint64(),
		tidbVersion, req.UseNewCdcKernel) {
		isNewKernelEnabled = true
		logger.Info("before override version and config for new kernel", zap.Bool("useNewKernel", req.UseNewCdcKernel),
			zap.Any("ticdcClusterEnt", ticdcClusterEnt))
		if err := m.overrideVersionAndConfigForNewKernel(ctx, ticdcClusterEnt, tidbVersion); err != nil {
			logger.Error("failed to override version and config for new kernel",
				zap.Any("ticdcClusterEnt", ticdcClusterEnt), zap.Error(err))
			return err.Chain("override version and config for new kernel")
		}
		logger.Info("after override version and config for new kernel", zap.Any("ticdcClusterEnt", ticdcClusterEnt))
	} else {
		logger.Info("before save and config for old kernel", zap.Any("ticdcClusterEnt", ticdcClusterEnt))
		if err := m.saveAndConfigForOldKernel(ctx, ticdcClusterEnt); err != nil {
			logger.Error("failed to save and config for old kernel", zap.Any("ticdcClusterEnt", ticdcClusterEnt), zap.Error(err))
			return err.Chain("save and config for old kernel")
		}
		logger.Info("after save and config for old kernel", zap.Any("ticdcClusterEnt", ticdcClusterEnt))
	}
	// get config from config center
	configResp, err := m.configManager.Get(ctx, &domain.GetTiCDCClusterConfigReq{
		TiCDCClusterID: ticdcClusterEnt.ID.Uint64(),
	})
	if err != nil {
		return err.Chain("get ticdc cluster configuration")
	}
	logger.Info("get ticdc cluster configuration", zap.Any("configResp", configResp))
	instanceTypeResp, err := m.instanceTypeModel.GetCDCInstanceType(ctx, &domain.GetCDCInstanceTypeReq{
		Name: ticdcClusterEnt.SpecificationCode,
		CloudProvider: domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(ticdcClusterEnt.InfraProviderType)),
		Region: ticdcClusterEnt.Region,
	})
	if err != nil {
		return err.Chain("get ticdc instance type")
	}
	if isNewKernelEnabled {
		logger.Info("before overwrite instance type iops and throughput",
			zap.Any("instanceTypeResp", instanceTypeResp))
		m.overwriteInstanceTypeIOPSAndThroughput(instanceTypeResp.InstanceType)
		logger.Info("after overwrite instance type iops and throughput", zap.Any("instanceTypeResp", instanceTypeResp))
	}
	if err := m.createRegionalTiCDCClusterIgnoreExists(
		ctx, ticdcClusterEnt, configResp.Configuration, instanceTypeResp.InstanceType); err != nil {
		return err.Chain("create regional ticdc cluster")
	}
	return nil
}

func (m *clusterModelRegionalImpl) isNewKernelEnabled(
	tenantID, projectID, tidbClusterID uint64,
	tidbVersion *utils.Version,
	migrationUseNewKernel bool,
) bool {
	minNewTiCDCkernelCompatibleTiDBVersion := config.Global().ChangefeedConfig.MinNewkernelCompatibleTiDBVersion
	lt, _ := tidbVersion.Lt(minNewTiCDCkernelCompatibleTiDBVersion)
	if lt {
		return false
	}
	if migrationUseNewKernel {
		return true
	}

	featureGateReq := &feature_gate.QueryFeatureGateReq{
		TenantID:  tenantID,
		ProjectID: projectID,
		ClusterID: tidbClusterID,
	}
	inBlacklist := m.featureGate.IsFeatureEnabled(featureGateReq, feature_gate.TiCDCNewKernelBlacklist)
	return !inBlacklist
}

func (m *clusterModelRegionalImpl) overrideVersionAndConfigForNewKernel(
	ctx context.Context,
	ticdcClusterEnt *entity.TiCDCCluster,
	tidbVersion *utils.Version,
) bizerr.BizError {
	minNewTiCDCkernelVersion := config.Global().ChangefeedConfig.MinNewkernelVersion
	lt, _ := tidbVersion.Lt(minNewTiCDCkernelVersion)
	if lt {
		// override version
		ticdcClusterEnt.Version = minNewTiCDCkernelVersion
	}
	ticdcClusterEnt.IsNewKernel = true
	// save records
	if err := m.ticdcClusterRepo.Save(ctx, ticdcClusterEnt); err != nil {
		return err.Chain("save ticdc cluster entity")
	}
	// init config in config center
	if _, err := m.configManager.Create(ctx, &domain.CreateTiCDCClusterConfigReq{
		TenantID:           ticdcClusterEnt.TenantID.Uint64(),
		TiCDCClusterID:     ticdcClusterEnt.ID.Uint64(),
		TiDBClusterVersion: ticdcClusterEnt.Version,
		CloudProvider: domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(ticdcClusterEnt.InfraProviderType)),
	}); err != nil {
		return err.Chain("create ticdc cluster configuration")
	}

	if err := m.configManager.PatchConfig(ctx, &domain.PatchTiCDCClusterConfigReq{
		TiCDCClusterID: ticdcClusterEnt.ID.Uint64(),
		CustomParams: map[string]*dataflow.ParamItem{
			"newarch": {Value: structpb.NewBoolValue(true)},
		},
	}); err != nil {
		return err.Chain("add new kernel config to ticdc cluster")
	}
	return nil
}

func (m *clusterModelRegionalImpl) saveAndConfigForOldKernel(
	ctx context.Context,
	ticdcClusterEnt *entity.TiCDCCluster,
) bizerr.BizError {
	// save records
	if err := m.ticdcClusterRepo.Save(ctx, ticdcClusterEnt); err != nil {
		return err.Chain("save ticdc cluster entity")
	}
	// init config in config center
	if _, err := m.configManager.Create(ctx, &domain.CreateTiCDCClusterConfigReq{
		TenantID:           ticdcClusterEnt.TenantID.Uint64(),
		TiCDCClusterID:     ticdcClusterEnt.ID.Uint64(),
		TiDBClusterVersion: ticdcClusterEnt.Version,
		CloudProvider: domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(ticdcClusterEnt.InfraProviderType)),
	}); err != nil {
		return err.Chain("create ticdc cluster configuration")
	}
	return nil
}

func (m *clusterModelRegionalImpl) createRegionalTiCDCClusterIgnoreExists(
	ctx context.Context,
	ticdcClusterEnt *entity.TiCDCCluster,
	config *domain.TiCDCConfigData,
	instanceType *domain.CDCInstanceType,
) bizerr.BizError {
	ticdcCluster, err := m.convertRegionalTiCDCCluster(ticdcClusterEnt, config, instanceType)
	if err != nil {
		return err.Chain("convert regional ticdc cluster")
	}
	region := &regionalapi.Region{
		TenantID: ticdcClusterEnt.TenantID.Uint64(),
		Name:     ticdcClusterEnt.Region,
		CloudProvider: string(domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(ticdcClusterEnt.InfraProviderType))),
	}
	regionalReq := &ticdc_v1.CreateTiCDCClusterRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		TicdcCluster:   ticdcCluster,
	}
	if _, err := m.ticdcClusterCli.CreateTiCDCCluster(ctx, region, regionalReq); err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordExist {
			return nil
		}
		return err
	}
	return nil
}

func (m *clusterModelRegionalImpl) convertRegionalTiCDCCluster(
	ticdcClusterEnt *entity.TiCDCCluster, config *domain.TiCDCConfigData, instanceType *domain.CDCInstanceType,
) (*ticdc_v1.TiCDCCluster, bizerr.BizError) {
	ticdcClusterConfig, err := m.convertRegionalTiCDCClusterConfig(config)
	if err != nil {
		return nil, err
	}
	ticdcCluster := &ticdc_v1.TiCDCCluster{
		Id:               ticdcClusterEnt.Name,
		TenantId:         strconv.FormatUint(ticdcClusterEnt.TenantID.Uint64(), 10),
		ProjectId:        strconv.FormatUint(ticdcClusterEnt.ProjectID.Uint64(), 10),
		ClusterId:        strconv.FormatUint(ticdcClusterEnt.ClusterID.Uint64(), 10),
		NodeInstanceType: instanceType.MachineType,
		NodeCount:        instanceType.Replicas,
		Storage: &ticdc_v1.TiCDCCluster_Storage{
			Type:          instanceType.StorageType,
			SizeGi:        instanceType.MachineStorageGi,
			Iops:          instanceType.IOPS,
			ThroughputMbs: instanceType.ThroughputMbs,
		},
		Configuration: ticdcClusterConfig,
	}
	return ticdcCluster, nil
}

func (m *clusterModelRegionalImpl) convertRegionalTiCDCClusterConfig(
	configData *domain.TiCDCConfigData,
) (*ticdc_v1.TiCDCCluster_Configuration, bizerr.BizError) {
	configBytes, err := toml.Marshal(configData.Config)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "marshal config")
	}
	configuration := &ticdc_v1.TiCDCCluster_Configuration{
		Config: string(configBytes),
	}
	if configData.BaseImage != nil {
		configuration.BaseImage = *configData.BaseImage
	}
	if configData.Tag != nil {
		configuration.Tag = *configData.Tag
	}
	return configuration, nil
}

func (m *clusterModelRegionalImpl) createReq2Entity(
	req *domain.CreateTiCDCClusterReq, logicalRegionName string,
) *entity.TiCDCCluster {
	return &entity.TiCDCCluster{
		InfraProviderType: req.InfraProviderType,
		CloudAccountID:    req.CloudAccountID,

		ClusterID: entity.ID(req.ClusterID),
		ProjectID: entity.ID(req.ProjectID),
		TenantID:  entity.ID(req.TenantID),

		Name:              req.Name,
		BusinessType:      ticdcClusterBizType2Entity(req.BizType),
		Region:            req.Region,
		LogicalRegion:     logicalRegionName,
		SpecificationCode: req.SpecificationCode,
		Status:            entity.TiCDCClusterStatusCreating,
		Version:           req.Version,
		DeployType:        req.DeployType,
		IsRegional:        true,
	}
}

func (m *clusterModelRegionalImpl) Scale(
	ctx context.Context, req *domain.ScaleTiCDCClusterReq,
) (*domain.ScaleTiCDCClusterResp, bizerr.BizError) {
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, req.ID)
	if err != nil {
		return nil, err.Chain("load ticdc cluster entity")
	}
	instanceTypeResp, err := m.instanceTypeModel.GetCDCInstanceType(ctx, &domain.GetCDCInstanceTypeReq{
		Name: req.SpecificationCode,
		CloudProvider: domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(ticdcClusterEnt.InfraProviderType)),
		Region: ticdcClusterEnt.Region,
	})
	if err != nil {
		return nil, err.Chain("get ticdc instance type")
	}

	if ticdcClusterEnt.IsNewKernel {
		m.overwriteInstanceTypeIOPSAndThroughput(instanceTypeResp.InstanceType)
	}
	customStorage, err := m.ticdcCustomStorageRepo.LoadByTiCDCClusterID(ctx, req.ID)
	if err != nil && err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return nil, bizerr.WrapError(err, "load ticdc custom storage from db failed")
	}
	// overwrite instance type if custom storage configuration exists
	if customStorage != nil {
		logger := log.GetLogger(ctx)
		logger.Info("ticdc custom storage exist, overwrite!",
			zap.Uint64("ticdc cluster id", ticdcClusterEnt.ID.Uint64()),
			zap.String("ticdc cluster name", ticdcClusterEnt.Name),
			zap.Any("instance type", instanceTypeResp.InstanceType),
			zap.Any("custom storage", customStorage))
		instanceTypeResp.InstanceType.IOPS = customStorage.Iops
		instanceTypeResp.InstanceType.ThroughputMbs = customStorage.ThroughputMbs
		instanceTypeResp.InstanceType.MachineStorageGi = customStorage.SizeGi
	}
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.doScale(ctx, req, ticdcClusterEnt, instanceTypeResp.InstanceType)
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	return &domain.ScaleTiCDCClusterResp{}, nil
}

func (m *clusterModelRegionalImpl) doScale(
	ctx context.Context,
	req *domain.ScaleTiCDCClusterReq,
	ticdcClusterEnt *entity.TiCDCCluster,
	instanceType *domain.CDCInstanceType,
) bizerr.BizError {
	ticdcClusterEnt.SpecificationCode = req.SpecificationCode
	ticdcClusterEnt.Status = entity.TiCDCClusterStatusScaling
	if err := m.ticdcClusterRepo.Save(ctx, ticdcClusterEnt); err != nil {
		return err.Chain("save ticdc cluster entity")
	}
	return m.scaleRegionalTiCDCCluster(ctx, ticdcClusterEnt, instanceType)
}

func (m *clusterModelRegionalImpl) scaleRegionalTiCDCCluster(
	ctx context.Context, ticdcClusterEnt *entity.TiCDCCluster, instanceType *domain.CDCInstanceType,
) bizerr.BizError {
	region := &regionalapi.Region{
		TenantID: ticdcClusterEnt.TenantID.Uint64(),
		Name:     ticdcClusterEnt.Region,
		CloudProvider: string(domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(ticdcClusterEnt.InfraProviderType))),
	}
	regionalReq := &ticdc_v1.ScaleTiCDCClusterRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		InstanceType:   instanceType.MachineType,
		Count:          instanceType.Replicas,
		Storage: &ticdc_v1.TiCDCCluster_Storage{
			Type:          instanceType.StorageType,
			SizeGi:        instanceType.MachineStorageGi,
			Iops:          instanceType.IOPS,
			ThroughputMbs: instanceType.ThroughputMbs,
		},
	}
	return m.ticdcClusterCli.ScaleTiCDCCluster(ctx, region, regionalReq)
}

func (m *clusterModelRegionalImpl) Query(
	ctx context.Context, req *domain.QueryTiCDCClusterReq,
) (*domain.QueryTiCDCClusterResp, bizerr.BizError) {
	ticdcClusterEnt, err := m.ticdcClusterRepo.Get(ctx, genFilterByQueryTiCDCClusterReq(req))
	if err != nil {
		return nil, err.Chain("get ticdc cluster entity")
	}
	return &domain.QueryTiCDCClusterResp{
		TiCDCClusterEntity: ticdcClusterEnt,
	}, nil
}

func (m *clusterModelRegionalImpl) Delete(
	ctx context.Context, req *domain.DeleteTiCDCClusterReq,
) (*domain.DeleteTiCDCClusterResp, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, req.ID)
	if err != nil && err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return nil, err.Chain("load ticdc cluster entity")
	}
	if ticdcClusterEnt == nil {
		logger.Info("ticdc cluster is not found", zap.Uint64("cluster_id", req.ID))
		return &domain.DeleteTiCDCClusterResp{}, nil
	}
	if ticdcClusterEnt.Status == entity.TiCDCClusterStatusDeleting ||
		ticdcClusterEnt.Status == entity.TiCDCClusterStatusDeleted {
		logger.Info("ticdc cluster is deleted",
			zap.Uint64("cluster_id", req.ID), zap.String("status", string(ticdcClusterEnt.Status)))
		return &domain.DeleteTiCDCClusterResp{}, nil
	}

	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.doDelete(ctx, ticdcClusterEnt)
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	return &domain.DeleteTiCDCClusterResp{}, nil
}

func (m *clusterModelRegionalImpl) doDelete(ctx context.Context, ticdcClusterEnt *entity.TiCDCCluster) bizerr.BizError {
	ticdcClusterEnt.Status = entity.TiCDCClusterStatusDeleting
	if err := m.ticdcClusterRepo.Save(ctx, ticdcClusterEnt); err != nil {
		return err.Chain("save ticdc cluster entity")
	}
	return m.deleteRegionalTiCDCCluster(ctx, ticdcClusterEnt)
}

func (m *clusterModelRegionalImpl) deleteRegionalTiCDCCluster(
	ctx context.Context, ticdcClusterEnt *entity.TiCDCCluster,
) bizerr.BizError {
	region := &regionalapi.Region{
		TenantID: ticdcClusterEnt.TenantID.Uint64(),
		Name:     ticdcClusterEnt.Region,
		CloudProvider: string(domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(ticdcClusterEnt.InfraProviderType))),
	}
	regionalReq := &ticdc_v1.DeleteTiCDCClusterRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
	}
	return m.ticdcClusterCli.DeleteTiCDCCluster(ctx, region, regionalReq)
}

func (m *clusterModelRegionalImpl) ApplyConfig(
	ctx context.Context, req *domain.ApplyTiCDCClusterConfigReq,
) bizerr.BizError {
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, req.TiCDCClusterID)
	if err != nil {
		return err.Chain("load ticdc cluster entity")
	}
	configResp, err := m.configManager.Get(ctx, &domain.GetTiCDCClusterConfigReq{
		TiCDCClusterID: req.TiCDCClusterID,
	})
	if err != nil {
		return err.Chain("get ticdc cluster configuration")
	}

	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.updateConfigRegionalTiCDCCluster(ctx, ticdcClusterEnt, configResp.Configuration)
	})
	return convertTransactionError(transErr)
}

func (m *clusterModelRegionalImpl) updateConfigRegionalTiCDCCluster(
	ctx context.Context, ticdcClusterEnt *entity.TiCDCCluster, config *domain.TiCDCConfigData,
) bizerr.BizError {
	ticdcClusterConfig, err := m.convertRegionalTiCDCClusterConfig(config)
	if err != nil {
		return err.Chain("convert regional ticdc cluster config")
	}

	ticdcClusterEnt.Status = entity.TiCDCClusterStatusUpgrading
	if err := m.ticdcClusterRepo.Save(ctx, ticdcClusterEnt); err != nil {
		return err.Chain("save ticdc cluster entity")
	}

	region := &regionalapi.Region{
		TenantID: ticdcClusterEnt.TenantID.Uint64(),
		Name:     ticdcClusterEnt.Region,
		CloudProvider: string(domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(ticdcClusterEnt.InfraProviderType))),
	}
	regionalReq := &ticdc_v1.UpdateTiCDCClusterConfigRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		Configuration:  ticdcClusterConfig,
	}
	return m.ticdcClusterCli.UpdateTiCDCClusterConfig(ctx, region, regionalReq)
}

func (m *clusterModelRegionalImpl) UpdateStorage(
	ctx context.Context, req *domain.UpdateTiCDCClusterStorageReq,
) bizerr.BizError {
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, req.TiCDCClusterID)
	if err != nil {
		return err.Chain("load ticdc cluster entity")
	}
	instanceTypeResp, err := m.instanceTypeModel.GetCDCInstanceType(ctx, &domain.GetCDCInstanceTypeReq{
		Name: ticdcClusterEnt.SpecificationCode,
		CloudProvider: domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(ticdcClusterEnt.InfraProviderType)),
		Region: ticdcClusterEnt.Region,
	})
	if err != nil {
		return err.Chain("get ticdc instance type")
	}
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.updateStorage(ctx, req, ticdcClusterEnt, instanceTypeResp)
	})
	return convertTransactionError(transErr)
}

func (m *clusterModelRegionalImpl) updateStorage(ctx context.Context,
	req *domain.UpdateTiCDCClusterStorageReq,
	ticdcClusterEnt *entity.TiCDCCluster,
	instanceTypeResp *domain.GetCDCInstanceTypeResp,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	if req.SizeGi == 0 && req.Iops == 0 && req.ThroughputMbs == 0 {
		logger.Info("no storage update", zap.Uint64("cluster_id", req.TiCDCClusterID))
		return nil
	}
	if req.SizeGi <= 0 {
		logger.Info("no storage size update",
			zap.Uint64("cluster_id", req.TiCDCClusterID),
			zap.Uint32("size_gi", req.SizeGi))
		req.SizeGi = instanceTypeResp.InstanceType.MachineStorageGi
	}
	if req.Iops <= 0 {
		logger.Info("no storage iops update",
			zap.Uint64("cluster_id", req.TiCDCClusterID),
			zap.Uint32("iops", req.Iops))
		req.Iops = instanceTypeResp.InstanceType.IOPS
	}
	if req.ThroughputMbs <= 0 {
		logger.Info("no storage throughput update",
			zap.Uint64("cluster_id", req.TiCDCClusterID),
			zap.Uint32("throughput_mbs", req.ThroughputMbs))
		req.ThroughputMbs = instanceTypeResp.InstanceType.ThroughputMbs
	}
	customStorage := &entity.TiCDCCustomStorage{
		TiCDCClusterID: entity.ID(req.TiCDCClusterID),
		SizeGi:         req.SizeGi,
		Iops:           req.Iops,
		ThroughputMbs:  req.ThroughputMbs,
	}

	if err := m.ticdcCustomStorageRepo.Save(ctx, customStorage); err != nil {
		return err.Chain("save ticdc custom storage entity")
	}

	region := &regionalapi.Region{
		TenantID: ticdcClusterEnt.TenantID.Uint64(),
		Name:     ticdcClusterEnt.Region,
		CloudProvider: string(domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(ticdcClusterEnt.InfraProviderType))),
	}
	regionalReq := &ticdc_v1.UpdateTiCDCClusterStorageRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		Storage: &ticdc_v1.TiCDCCluster_Storage{
			Type:          instanceTypeResp.InstanceType.StorageType,
			SizeGi:        customStorage.SizeGi,
			Iops:          customStorage.Iops,
			ThroughputMbs: customStorage.ThroughputMbs,
		},
	}
	return m.ticdcClusterCli.UpdateTiCDCClusterStorage(ctx, region, regionalReq)
}

func (m *clusterModelRegionalImpl) List(
	ctx context.Context, req *domain.ListTiCDCClustersReq,
) (*domain.ListTiCDCClustersResp, bizerr.BizError) {
	ticdcClusterEnts, err := m.ticdcClusterRepo.List(
		ctx, func(qs entity.TiCDCClusterQuerySet) entity.TiCDCClusterQuerySet {
			return qs.ClusterIDEq(entity.ID(req.TiDBClusterID)).DeletedAtIsNull()
		})
	if err != nil {
		return nil, err.Chain("list ticdc clusters")
	}
	return &domain.ListTiCDCClustersResp{
		Items: ticdcClusterEnts,
	}, nil
}

func (m *clusterModelRegionalImpl) Restart(ctx context.Context, req *domain.RestartTiCDCClusterReq) bizerr.BizError {
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, req.TiCDCClusterID)
	if err != nil {
		return err.Chain("load ticdc cluster entity")
	}
	region := &regionalapi.Region{
		TenantID: ticdcClusterEnt.TenantID.Uint64(),
		Name:     ticdcClusterEnt.Region,
		CloudProvider: string(domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(ticdcClusterEnt.InfraProviderType))),
	}
	regionalReq := &ticdc_v1.RestartTiCDCClusterRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
	}
	if err := m.ticdcClusterCli.RestartTiCDCCluster(ctx, region, regionalReq); err != nil {
		return err.Chain("restart ticdc cluster")
	}
	log.GetLogger(ctx).Info(
		"restart ticdc cluster successfully", zap.Uint64("ticdc_cluster_id", req.TiCDCClusterID))
	return nil
}

func (m *clusterModelRegionalImpl) GetRole(
	ctx context.Context, req *domain.GetTiCDCClusterRoleReq,
) (*domain.GetTiCDCClusterRoleResp, bizerr.BizError) {
	region := &regionalapi.Region{
		TenantID:      req.TenantID,
		Name:          req.Region,
		CloudProvider: string(req.CloudProvider),
	}
	getRoleReq := &ticdc_v1.GetTiCDCClusterRoleRequest{
		TenantId:      strconv.FormatUint(req.TenantID, 10),
		ProjectId:     strconv.FormatUint(req.ProjectID, 10),
		TidbClusterId: strconv.FormatUint(req.TiDBClusterID, 10),
	}
	roleInfo, err := m.ticdcClusterCli.GetTiCDCClusterRole(ctx, region, getRoleReq)
	if err != nil {
		return nil, err.Chain("get ticdc cluster role")
	}
	return &domain.GetTiCDCClusterRoleResp{
		CloudAccountID:   roleInfo.CloudAccountId,
		CloudIAMIdentity: roleInfo.CloudIamIdentity,
		AWSExternalID:    roleInfo.AwsExternalId,
	}, nil
}

func (m *clusterModelRegionalImpl) GetExistingClusterVersion(
	ctx context.Context, req *domain.GetExistingTiCDCClusterVersionReq,
) (*domain.GetExistingTiCDCClusterVersionResp, bizerr.BizError) {
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, req.TiCDCClusterID)
	if err != nil {
		return nil, err.Chain("load ticdc cluster entity")
	}
	if ticdcClusterEnt.Status == entity.TiCDCClusterStatusDeleting ||
		ticdcClusterEnt.Status == entity.TiCDCClusterStatusDeleted {
		return nil, bizerr.New(bizerr.ErrCodeParams, "ticdc cluster is deleted")
	}

	if !ticdcClusterEnt.IsNewKernel {
		return &domain.GetExistingTiCDCClusterVersionResp{
			Version: ticdcClusterEnt.Version,
			NewArch: false,
		}, nil
	}

	// for ticdc new kernel, the initial version of ticdc cluster entity is always v9.0.0-beta.1
	// we need to get the real version from the image tag
	region := &regionalapi.Region{
		TenantID: ticdcClusterEnt.TenantID.Uint64(),
		Name:     ticdcClusterEnt.Region,
		CloudProvider: string(domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(ticdcClusterEnt.InfraProviderType))),
	}
	getReq := &ticdc_v1.GetTiCDCClusterRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
	}
	ticdcCluster, err := m.ticdcClusterCli.GetTiCDCCluster(ctx, region, getReq)
	if err != nil {
		return nil, err.Chain("get regional ticdc cluster")
	}

	return &domain.GetExistingTiCDCClusterVersionResp{
		Version: ticdcCluster.Configuration.Tag,
		NewArch: true,
	}, nil
}

func (m *clusterModelRegionalImpl) GetNewClusterVersion(
	ctx context.Context, req *domain.GetNewTiCDCClusterVersionReq,
) (*domain.GetNewTiCDCClusterVersionResp, bizerr.BizError) {
	tidbVersion, verr := utils.NewVersion(req.TiDBClusterVersion)
	if verr != nil {
		return nil, bizerr.WrapErrorWithCode(
			verr, bizerr.ErrCodeParams, "tidb version is invalid: %s", req.TiDBClusterVersion)
	}

	newArch := m.isNewKernelEnabled(req.TenantID, req.ProjectID, req.TiDBClusterID, tidbVersion, false)
	if !newArch {
		return &domain.GetNewTiCDCClusterVersionResp{
			Version: req.TiDBClusterVersion,
			NewArch: false,
		}, nil
	}

	// for ticdc new kernel, the real version is the tag of version v9.0.0-beta.1
	imageReq := &domain.GetNewTiCDCClusterImageReq{
		TiCDCClusterVersion: req.TiDBClusterVersion,
	}
	minNewTiCDCKernelVersion := config.Global().ChangefeedConfig.MinNewkernelVersion
	lt, _ := tidbVersion.Lt(minNewTiCDCKernelVersion)
	if lt {
		// override version
		imageReq.TiCDCClusterVersion = minNewTiCDCKernelVersion
	}
	imageResp, err := m.configManager.GetNewClusterImage(ctx, imageReq)
	if err != nil {
		return nil, err.Chain("get new ticdc cluster image")
	}

	return &domain.GetNewTiCDCClusterVersionResp{
		Version: imageResp.Tag,
		NewArch: true,
	}, nil
}

func convertTransactionError(err error) bizerr.BizError {
	if err == nil {
		return nil
	}

	var bizErr bizerr.BizError
	if errors.As(err, &bizErr) {
		return bizErr
	}
	return bizerr.New(bizerr.ErrCodeDB, "execute transaction error: %s", err.Error())
}

func ticdcClusterBizType2Entity(bizType domain.TiCDCClusterBizType) entity.TiCDCClusterBizType {
	switch bizType {
	case domain.TiCDCClusterBizTypeRecoveryGroup:
		return entity.TiCDCClusterBizTypeRecoveryGroup
	case domain.TiCDCClusterBizTypeChangefeed:
		return entity.TiCDCClusterBizTypeChangefeed
	default:
		return entity.TiCDCClusterBizTypeChangefeed
	}
}

func (m *clusterModelRegionalImpl) overwriteInstanceTypeIOPSAndThroughput(instanceType *domain.CDCInstanceType) {
	if spec, exists := newKernelCPUCore2DiskIOMap[instanceType.MachineCPUNum]; exists {
		instanceType.IOPS = spec.IOPS
		instanceType.ThroughputMbs = spec.ThroughputMbs
	}
}

func genFilterByQueryTiCDCClusterReq(req *domain.QueryTiCDCClusterReq) repo.TiCDCClusterFilterOp {
	return func(qs entity.TiCDCClusterQuerySet) entity.TiCDCClusterQuerySet {
		if req.ID != 0 {
			qs = qs.IDEq(entity.ID(req.ID))
		}
		if req.Name != "" {
			qs = qs.NameEq(req.Name).DeletedAtIsNull()
		}
		return qs
	}
}