package clusterconfig

import (
	"context"
	"strconv"

	"github.com/samber/lo"
	configcenter "github.com/tidbcloud/config-center/proto/config_center"
	regionaldedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/proto"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/enum"
)

const InitSql = "INIT_SQL"

func (cc *configClientImpl) GetClusterVersion(ctx context.Context, cluster *entity.Cluster) (string, bizerr.BizError) {
	// till cluster normal use config_center
	resp, err := cc.configCenterCli.GetClusterMeta(ctx, &configcenter.GetClusterMetaRequest{
		ClusterType:   configcenter.ClusterType_DEDICATED_TIDB,
		ClusterId:     strconv.FormatUint(cluster.ClusterId, 10),
		MainComponent: globaldedicatedv1.ComponentType_TIDB.String(),
	})

	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			// not found, use default version
			if s.Code() == codes.NotFound {
				return cc.getDefaultVersion(ctx, cluster)
			}
			return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "get cluster meta: %s", s.Message())
		}
		return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "get cluster meta: %s", err.Error())
	}

	if resp.Version == "" {
		return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "cluster meta version is empty")
	}

	return resp.Version, nil
}

func (cc *configClientImpl) getDefaultVersion(ctx context.Context, cluster *entity.Cluster) (string, bizerr.BizError) {
	resp, err := cc.configCenterCli.GetDefaultVersion(ctx, &configcenter.GetDefaultVersionRequest{
		ClusterType:   configcenter.ClusterType_DEDICATED_TIDB,
		Component:     globaldedicatedv1.ComponentType_TIDB.String(),
		CloudProvider: proto.String(cluster.CloudProviderId.String()),
		Region:        proto.String(cluster.RegionId.String()),
		TenantId:      proto.Uint64(cluster.OrganizationId),
	})
	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "get cluster default version: %s", s.Message())
		}
		return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "get cluster default version: %s", err.Error())
	}

	if resp.Version == "" {
		return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "cluster default version is empty")
	}

	return resp.Version, nil
}

func (cc *configClientImpl) GetTiproxyVersion(ctx context.Context, cluster *entity.Cluster) (string, bizerr.BizError) {
	// till cluster normal use config_center
	resp, err := cc.configCenterCli.GetClusterComponent(ctx, &configcenter.GetClusterComponentRequest{
		ClusterType:   configcenter.ClusterType_DEDICATED_TIDB,
		ClusterId:     strconv.FormatUint(cluster.ClusterId, 10),
		ComponentName: globaldedicatedv1.ComponentType_TIPROXY.String(),
	})

	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			// not found, use default version
			if s.Code() == codes.NotFound {
				return cc.getTiProxyDefaultVersion(ctx, cluster)
			}
			return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "get tiroxy meta: %s", s.Message())
		}
		return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "get tiroxy meta: %s", err.Error())
	}

	if resp.Version == "" {
		return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "tiproxy meta version is empty")
	}

	return resp.Version, nil
}

func (cc *configClientImpl) getTiProxyDefaultVersion(
	ctx context.Context,
	cluster *entity.Cluster,
) (string, bizerr.BizError) {
	resp, err := cc.configCenterCli.GetDefaultVersion(ctx, &configcenter.GetDefaultVersionRequest{
		ClusterType:   configcenter.ClusterType_DEDICATED_TIDB,
		Component:     globaldedicatedv1.ComponentType_TIPROXY.String(),
		CloudProvider: proto.String(cluster.CloudProviderId.String()),
		Region:        proto.String(cluster.RegionId.String()),
		TenantId:      proto.Uint64(cluster.OrganizationId),
	})
	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "get default tiproxy version: %s", s.Message())
		}
		return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "get default tiproxy version: %s", err.Error())
	}

	if resp.Version == "" {
		return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "tiproxy default version is empty")
	}

	return resp.Version, nil
}

func (cc *configClientImpl) CreateClusterMeta(ctx context.Context, cluster *entity.Cluster) (string, bizerr.BizError) {
	metaReq := &configcenter.AutoUpdateClusterMetaRequest{
		ClusterType: configcenter.ClusterType_DEDICATED_TIDB,
		ClusterId:   strconv.FormatUint(cluster.ClusterId, 10),
		Version:     proto.String(cluster.Version),
		ComponentRequests: make(
			[]*configcenter.UpdateClusterComponentRequest,
			0,
			len(cluster.ClusterComponents),
		),
		ConfigurationFileRequests: make(
			[]*configcenter.AutoUpdateClusterConfigurationFileRequest,
			0,
			len(cluster.ClusterComponents),
		),
	}

	// full component image and configuration
	for _, component := range cluster.ClusterComponents {
		version := cluster.Version

		if component.GetComponentType() == globaldedicatedv1.ComponentType_TIPROXY {
			var err bizerr.BizError
			version, err = cc.GetTiproxyVersion(ctx, cluster)
			if err != nil {
				return "", err
			}
		}

		metaReq.ConfigurationFileRequests = append(metaReq.ConfigurationFileRequests,
			&configcenter.AutoUpdateClusterConfigurationFileRequest{
				ConfigurationFilename: component.GetComponentType().String(),
				ConfigurationType:     proto.String(component.GetComponentType().String()),
				BelongToComponent:     proto.String(component.GetComponentType().String()),
				Format:                configcenter.Format_TOML.Enum(),
				CloudProvider:         cluster.CloudProviderId.String(),
				NodeInstance:          component.GetInstanceType(),
				TenantId:              proto.Uint64(cluster.OrganizationId),
				Version:               version,
			})

		metaReq.ComponentRequests = append(metaReq.ComponentRequests, &configcenter.UpdateClusterComponentRequest{
			ComponentName:     component.GetComponentType().String(),
			CloudProvider:     proto.String(cluster.CloudProviderId.String()),
			NodeInstance:      proto.String(component.GetInstanceType()),
			OverwriteImageTag: proto.String(""),
			AllowMissing:      true,
			Version:           lo.ToPtr(version),
		})
	}

	// check if the tiflash component is in the cluster
	if tiflash, ok := lo.Find(cluster.ClusterComponents, func(c *entity.ClusterComponent) bool {
		return c.GetComponentType() == globaldedicatedv1.ComponentType_TIFLASH
	}); ok {
		// if tiflash is in the cluster, create the tiflash learner component
		metaReq.ConfigurationFileRequests = append(metaReq.ConfigurationFileRequests,
			&configcenter.AutoUpdateClusterConfigurationFileRequest{
				ConfigurationFilename: globaldedicatedv1.ComponentType_TIFLASH_LEARNER.String(),
				ConfigurationType:     proto.String(globaldedicatedv1.ComponentType_TIFLASH_LEARNER.String()),
				BelongToComponent:     proto.String(globaldedicatedv1.ComponentType_TIFLASH.String()),
				Format:                configcenter.Format_TOML.Enum(),
				CloudProvider:         cluster.CloudProviderId.String(),
				NodeInstance:          tiflash.GetInstanceType(),
				TenantId:              proto.Uint64(cluster.OrganizationId),
				Version:               cluster.Version,
			})
	}

	// add init sql
	metaReq.ConfigurationFileRequests = append(metaReq.ConfigurationFileRequests,
		&configcenter.AutoUpdateClusterConfigurationFileRequest{
			ConfigurationFilename: InitSql, // "INIT_SQL"
			ConfigurationType:     proto.String(InitSql),
			Format:                configcenter.Format_LINES.Enum(),
			CloudProvider:         cluster.CloudProviderId.String(),
			NodeInstance:          "", // no instance type
			TenantId:              proto.Uint64(cluster.OrganizationId),
		})

	_, err := cc.configCenterCli.AutoUpdateClusterMeta(ctx, metaReq)

	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
		}
		return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", err.Error())
	}
	return cluster.Version, nil
}

func (cc *configClientImpl) GetClusterComponentsVersion(
	ctx context.Context,
	cluster *entity.Cluster,
) (*regionaldedicatedv1.ClusterVersion, bizerr.BizError) {
	clusterVersion := &regionaldedicatedv1.ClusterVersion{
		ClusterVersion: cluster.Version,
		ComponentVersions: make(
			[]*regionaldedicatedv1.ClusterVersion_ComponentVersion,
			0,
			len(cluster.ClusterComponents),
		),
	}

	resp, err := cc.configCenterCli.GetClusterMeta(ctx, &configcenter.GetClusterMetaRequest{
		ClusterType:   configcenter.ClusterType_DEDICATED_TIDB,
		ClusterId:     strconv.FormatUint(cluster.ClusterId, 10),
		MainComponent: globaldedicatedv1.ComponentType_TIDB.String(),
	})
	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
		}
		return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", err.Error())
	}

	for _, component := range resp.Components {
		componentType, err := cvtComponentType(component.ComponentName)
		if err != nil {
			tilog.Warn(
				ctx,
				"unknown component type",
				zap.String("componentType", component.ComponentName),
				zap.Uint64("clusterId", cluster.ClusterId),
			)
			continue
		}

		if componentType == regionaldedicatedv1.ComponentType_TIDB {
			clusterVersion.ClusterVersion = component.Version
		}

		// check repo and tag
		if component.BaseImage == "" || component.EffectiveImageTag == "" {
			tilog.Warn(
				ctx,
				"component image not found",
				zap.String("componentType", componentType.String()),
				zap.Uint64("clusterId", cluster.ClusterId),
			)
			return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "component image not found")
		}

		clusterVersion.ComponentVersions = append(
			clusterVersion.ComponentVersions,
			&regionaldedicatedv1.ClusterVersion_ComponentVersion{
				ComponentType: componentType,
				ImageRepo:     component.BaseImage,
				ImageTag:      component.EffectiveImageTag,
			},
		)
	}

	if clusterVersion.ClusterVersion != cluster.Version {
		tilog.Warn(
			ctx,
			"cluster version not match",
			zap.Uint64("clusterId", cluster.ClusterId),
			zap.String("expectedVersion", cluster.Version),
			zap.String("actualVersion", clusterVersion.ClusterVersion),
		)
		return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "cluster version not match")
	}

	return clusterVersion, nil
}

func (cc *configClientImpl) ListComponentConfigurations(
	ctx context.Context,
	cluster *entity.Cluster,
) ([]*regionaldedicatedv1.ComponentConfiguration, bizerr.BizError) {
	var configs []*regionaldedicatedv1.ComponentConfiguration

	resp, err := cc.configCenterCli.GetClusterMeta(ctx, &configcenter.GetClusterMetaRequest{
		ClusterType:   configcenter.ClusterType_DEDICATED_TIDB,
		ClusterId:     strconv.FormatUint(cluster.ClusterId, 10),
		MainComponent: globaldedicatedv1.ComponentType_TIDB.String(),
	})
	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
		}
		return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", err.Error())
	}

	for _, file := range resp.ConfigurationFiles {
		if file.ConfigurationFilename == InitSql {
			continue
		}

		componentType, err := cvtComponentType(file.ConfigurationFilename)
		if err != nil {
			tilog.Warn(
				ctx,
				"unknown component type",
				zap.String("componentType", file.ConfigurationFilename),
				zap.Uint64("clusterId", cluster.ClusterId),
			)
			continue
		}

		configs = append(configs, &regionaldedicatedv1.ComponentConfiguration{
			ComponentType:             componentType,
			ConfigurationFileContents: file.Text,
		})
	}

	return configs, nil
}

func (cc *configClientImpl) CreateOrUpdateComponentConfiguration(
	ctx context.Context,
	cluster *entity.Cluster,
	component *entity.ClusterComponent,
) ([]*regionaldedicatedv1.ComponentConfiguration, bizerr.BizError) {
	var configs []*regionaldedicatedv1.ComponentConfiguration

	version := cluster.Version

	if component.GetComponentType() == globaldedicatedv1.ComponentType_TIPROXY {
		var err bizerr.BizError
		version, err = cc.GetTiproxyVersion(ctx, cluster)
		if err != nil {
			return nil, err
		}
	}

	requests := []*configcenter.AutoUpdateClusterConfigurationFileRequest{
		{
			ClusterType:           configcenter.ClusterType_DEDICATED_TIDB,
			ClusterId:             strconv.FormatUint(component.ClusterId, 10),
			ConfigurationFilename: component.GetComponentType().String(),
			ConfigurationType:     proto.String(component.GetComponentType().String()),
			Format:                configcenter.Format_TOML.Enum(),
			Version:               version,
			CloudProvider:         cluster.CloudProviderId.String(),
			NodeInstance:          component.GetInstanceType(),
			TenantId:              proto.Uint64(cluster.OrganizationId),
		},
	}

	// add tiflash learner config
	if component.GetComponentType() == globaldedicatedv1.ComponentType_TIFLASH {
		requests = append(requests, &configcenter.AutoUpdateClusterConfigurationFileRequest{
			ClusterType:           configcenter.ClusterType_DEDICATED_TIDB,
			ClusterId:             strconv.FormatUint(component.ClusterId, 10),
			ConfigurationFilename: globaldedicatedv1.ComponentType_TIFLASH_LEARNER.String(),
			ConfigurationType:     proto.String(globaldedicatedv1.ComponentType_TIFLASH_LEARNER.String()),
			BelongToComponent:     proto.String(globaldedicatedv1.ComponentType_TIFLASH.String()),
			Format:                configcenter.Format_TOML.Enum(),
			Version:               version,
			CloudProvider:         cluster.CloudProviderId.String(),
			NodeInstance:          component.GetInstanceType(),
			TenantId:              proto.Uint64(cluster.OrganizationId),
		})
	}

	resp, err := cc.configCenterCli.BatchAutoUpdateClusterConfigurationFiles(
		ctx,
		&configcenter.BatchAutoUpdateClusterConfigurationFilesRequest{Requests: requests},
	)
	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
		}
	}

	for _, file := range resp.Files {
		componentType, err := cvtComponentType(file.ConfigurationFilename)
		if err != nil {
			tilog.Warn(
				ctx,
				"unknown component type",
				zap.String("componentType", file.ConfigurationFilename),
				zap.Uint64("clusterId", cluster.ClusterId),
			)
			continue
		}
		configs = append(configs, &regionaldedicatedv1.ComponentConfiguration{
			ComponentType:             componentType,
			ConfigurationFileContents: file.Text,
		})
	}

	return configs, nil
}

func (cc *configClientImpl) CreateOrUpdateComponentVersion(
	ctx context.Context,
	cluster *entity.Cluster,
	component *entity.ClusterComponent,
) (*regionaldedicatedv1.ClusterVersion_ComponentVersion, bizerr.BizError) {
	componentType, berr := cvtComponentType(component.GetComponentType().String())
	if berr != nil {
		return nil, berr
	}

	version := cluster.Version

	if component.GetComponentType() == globaldedicatedv1.ComponentType_TIPROXY {
		var err bizerr.BizError
		version, err = cc.GetTiproxyVersion(ctx, cluster)
		if err != nil {
			return nil, err
		}
	}

	resp, err := cc.configCenterCli.UpdateClusterComponent(ctx, &configcenter.UpdateClusterComponentRequest{
		ClusterType:       configcenter.ClusterType_DEDICATED_TIDB,
		ClusterId:         strconv.FormatUint(component.ClusterId, 10),
		Version:           lo.ToPtr(version),
		ComponentName:     component.GetComponentType().String(),
		CloudProvider:     proto.String(cluster.CloudProviderId.String()),
		NodeInstance:      proto.String(component.GetInstanceType()),
		OverwriteImageTag: proto.String(""),
		AllowMissing:      true,
	})
	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
		}
	}

	// check repo and tag
	if resp.BaseImage == "" || resp.EffectiveImageTag == "" {
		tilog.Warn(
			ctx,
			"component image not found",
			zap.String("componentType", componentType.String()),
			zap.Uint64("clusterId", cluster.ClusterId),
		)
		return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "component image not found")
	}

	componentVersion := &regionaldedicatedv1.ClusterVersion_ComponentVersion{
		ComponentType: componentType,
		ImageRepo:     resp.BaseImage,
		ImageTag:      resp.EffectiveImageTag,
	}

	return componentVersion, nil
}

func (cc *configClientImpl) GetClusterInitSqlStmts(
	ctx context.Context,
	cluster *entity.Cluster,
) (string, bizerr.BizError) {
	resp, err := cc.configCenterCli.GetClusterConfigurationFile(ctx, &configcenter.GetClusterConfigurationFileRequest{
		ClusterType:           configcenter.ClusterType_DEDICATED_TIDB,
		ClusterId:             strconv.FormatUint(cluster.ClusterId, 10),
		ConfigurationFilename: InitSql,
	})
	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
		}
		return "", bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", err.Error())
	}
	return resp.Text, nil
}

func (cc *configClientImpl) GetClusterComponentVersion(
	ctx context.Context,
	cluster *entity.Cluster,
	componentType entity.ComponentType,
) (*regionaldedicatedv1.ClusterVersion_ComponentVersion, bizerr.BizError) {
	resp, err := cc.configCenterCli.GetClusterComponent(ctx, &configcenter.GetClusterComponentRequest{
		ClusterType:   configcenter.ClusterType_DEDICATED_TIDB,
		ClusterId:     strconv.FormatUint(cluster.ClusterId, 10),
		ComponentName: componentType.String(),
	})
	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
		}
		return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", err.Error())
	}

	regionalComponentType, berr := cvtComponentType(componentType.String())
	if berr != nil {
		return nil, berr
	}

	// check repo and tag
	if resp.BaseImage == "" || resp.EffectiveImageTag == "" {
		tilog.Warn(
			ctx,
			"component image not found",
			zap.String("componentType", componentType.String()),
			zap.Uint64("clusterId", cluster.ClusterId),
		)
		return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "component image not found")
	}

	return &regionaldedicatedv1.ClusterVersion_ComponentVersion{
		ComponentType: regionalComponentType,
		ImageRepo:     resp.BaseImage,
		ImageTag:      resp.EffectiveImageTag,
	}, nil
}

func (cc *configClientImpl) GetComponentConfiguration(
	ctx context.Context,
	cluster *entity.Cluster,
	componentType entity.ComponentType,
) (*regionaldedicatedv1.ComponentConfiguration, bizerr.BizError) {
	resp, err := cc.configCenterCli.GetClusterConfigurationFile(ctx, &configcenter.GetClusterConfigurationFileRequest{
		ClusterType:           configcenter.ClusterType_DEDICATED_TIDB,
		ClusterId:             strconv.FormatUint(cluster.ClusterId, 10),
		ConfigurationFilename: componentType.String(),
	})
	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
		}
		return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", err.Error())
	}
	regionalComponentType, berr := cvtComponentType(componentType.String())
	if berr != nil {
		return nil, berr
	}

	return &regionaldedicatedv1.ComponentConfiguration{
		ComponentType:             regionalComponentType,
		ConfigurationFileContents: resp.Text,
	}, nil
}

func (cc *configClientImpl) UpdateClusterComponentConfigurations(
	ctx context.Context,
	cluster *entity.Cluster,
	clusterConfiguration *entity.ClusterConfiguration,
) bizerr.BizError {
	componentImageTagReq := &configcenter.BatchUpdateClusterComponentsRequest{}
	componentConfigurationReq := &configcenter.BatchUpsertClusterCustomParamsRequest{}

	// init req
	for _, componentCfg := range clusterConfiguration.GetComponentConfigurations() {
		// image tag
		if componentCfg.GetOverwriteImageTag() != "" {
			componentImageTagReq.Requests = append(
				componentImageTagReq.Requests,
				&configcenter.UpdateClusterComponentRequest{
					ClusterType:       configcenter.ClusterType_DEDICATED_TIDB,
					ClusterId:         strconv.FormatUint(cluster.ClusterId, 10),
					ComponentName:     componentCfg.ComponentType.String(),
					OverwriteImageTag: lo.EmptyableToPtr(componentCfg.OverwriteImageTag),
				},
			)
		}

		// configuration
		if componentCfg.GetCustomParams() != nil {
			customParams := make(map[string]*configcenter.ParamItem, len(componentCfg.GetCustomParams()))
			for k, item := range componentCfg.GetCustomParams() {
				customParams[k] = &configcenter.ParamItem{Value: item.GetValue(), Comment: item.GetComment()}
			}
			componentConfigurationReq.Requests = append(
				componentConfigurationReq.Requests,
				&configcenter.UpsertClusterCustomParamsRequest{
					ClusterType:           configcenter.ClusterType_DEDICATED_TIDB,
					ClusterId:             strconv.FormatUint(cluster.ClusterId, 10),
					ConfigurationFilename: componentCfg.ComponentType.String(),
					UpsertedParams:        customParams,
				},
			)
		}
	}

	// update image tag
	if len(componentImageTagReq.Requests) > 0 {
		_, err := cc.configCenterCli.BatchUpdateClusterComponents(ctx, componentImageTagReq)
		if err != nil {
			s, ok := status.FromError(err)
			if ok {
				return bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
			}
			return bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", err.Error())
		}
	}

	// update configuration
	if len(componentConfigurationReq.Requests) > 0 {
		_, err := cc.configCenterCli.BatchUpsertClusterCustomParams(ctx, componentConfigurationReq)
		if err != nil {
			s, ok := status.FromError(err)
			if ok {
				return bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
			}
			return bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", err.Error())
		}
	}

	return nil
}

func cvtComponentType(componentType string) (regionaldedicatedv1.ComponentType, bizerr.BizError) {
	globalComponentType, err := enum.FromString[globaldedicatedv1.ComponentType](componentType)
	if err == nil {
		switch globaldedicatedv1.ComponentType(globalComponentType) {
		case globaldedicatedv1.ComponentType_TIDB:
			return regionaldedicatedv1.ComponentType_TIDB, nil
		case globaldedicatedv1.ComponentType_TIKV:
			return regionaldedicatedv1.ComponentType_TIKV, nil
		case globaldedicatedv1.ComponentType_PD:
			return regionaldedicatedv1.ComponentType_PD, nil
		case globaldedicatedv1.ComponentType_TIFLASH:
			return regionaldedicatedv1.ComponentType_TIFLASH, nil
		case globaldedicatedv1.ComponentType_TIFLASH_LEARNER:
			return regionaldedicatedv1.ComponentType_TIFLASH_LEARNER, nil
		case globaldedicatedv1.ComponentType_TIPROXY:
			return regionaldedicatedv1.ComponentType_TIPROXY, nil
		}
	}
	return regionaldedicatedv1.ComponentType_COMPONENT_TYPE_UNSPECIFIED, bizerr.New(
		bizerr.ErrCodeUnknown,
		"unknown component type: %s",
		componentType,
	)
}

// GetCustomClusterConfiguration gets the custom params configuration from config center.
func (cc *configClientImpl) GetCustomClusterConfiguration(
	ctx context.Context,
	cluster *entity.Cluster,
) (*entity.ClusterConfiguration, bizerr.BizError) {
	clusterConfiguration := &entity.ClusterConfiguration{
		ComponentConfigurations: make([]*globaldedicatedv1.ClusterConfiguration_ComponentConfiguration, 0, len(cluster.ClusterComponents)),
	}

	resp, err := cc.configCenterCli.GetClusterMeta(ctx, &configcenter.GetClusterMetaRequest{
		ClusterType:   configcenter.ClusterType_DEDICATED_TIDB,
		ClusterId:     strconv.FormatUint(cluster.ClusterId, 10),
		MainComponent: globaldedicatedv1.ComponentType_TIDB.String(),
	})
	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
		}
		return nil, bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", err.Error())
	}

	componentOverwriteImageTags := lo.SliceToMap(
		resp.GetComponents(),
		func(c *configcenter.ClusterComponent) (string, string) {
			return c.ComponentName, c.GetOverwriteImageTag()
		},
	)

	for _, file := range resp.ConfigurationFiles {
		if file.ConfigurationFilename == InitSql {
			continue
		}

		globalComponentType, err := enum.FromString[globaldedicatedv1.ComponentType](file.ConfigurationFilename)
		if err != nil {
			tilog.Warn(
				ctx,
				"unknown component type",
				zap.String("componentType", file.ConfigurationFilename),
				zap.Uint64("clusterId", cluster.ClusterId),
			)
			continue
		}

		customParams := make(map[string]*globaldedicatedv1.ClusterConfiguration_ComponentConfiguration_ParamItem)
		for k, v := range file.CustomParams {
			customParams[k] = &globaldedicatedv1.ClusterConfiguration_ComponentConfiguration_ParamItem{
				Value:   v.GetValue(),
				Comment: v.GetComment(),
			}
		}

		clusterConfiguration.ComponentConfigurations = append(
			clusterConfiguration.ComponentConfigurations,
			&globaldedicatedv1.ClusterConfiguration_ComponentConfiguration{
				ComponentType:     globaldedicatedv1.ComponentType(globalComponentType),
				CustomParams:      customParams,
				OverwriteImageTag: componentOverwriteImageTags[file.ConfigurationFilename],
			},
		)
	}

	return clusterConfiguration, nil
}

func (cc *configClientImpl) OverwriteClusterComponentConfigurations(
	ctx context.Context,
	cluster *entity.Cluster,
	clusterConfiguration *entity.ClusterConfiguration,
) bizerr.BizError {
	componentImageTagReq := &configcenter.BatchUpdateClusterComponentsRequest{}
	componentConfigurationReq := &configcenter.BatchUpdateClusterConfigurationFilesRequest{}

	existingComponentTypes := lo.SliceToMap(
		cluster.ClusterComponents,
		func(c *entity.ClusterComponent) (string, struct{}) {
			return c.GetComponentType().String(), struct{}{}
		},
	)
	// If tiflash exists, add tiflash learner. As the learner config is stored separately in config center.
	_, ok := existingComponentTypes[globaldedicatedv1.ComponentType_TIFLASH.String()]
	if ok {
		existingComponentTypes[globaldedicatedv1.ComponentType_TIFLASH_LEARNER.String()] = struct{}{}
	}

	// init req
	for _, componentCfg := range clusterConfiguration.GetComponentConfigurations() {
		// skip image tag and custom params for non-existing components
		if _, ok := existingComponentTypes[componentCfg.ComponentType.String()]; !ok {
			continue
		}

		// image tag
		if componentCfg.GetOverwriteImageTag() != "" {
			componentImageTagReq.Requests = append(
				componentImageTagReq.Requests,
				&configcenter.UpdateClusterComponentRequest{
					ClusterType:       configcenter.ClusterType_DEDICATED_TIDB,
					ClusterId:         strconv.FormatUint(cluster.ClusterId, 10),
					ComponentName:     componentCfg.ComponentType.String(),
					OverwriteImageTag: lo.EmptyableToPtr(componentCfg.OverwriteImageTag),
				},
			)
		}

		// configuration
		customParams := make(map[string]*configcenter.ParamItem)
		if componentCfg.GetCustomParams() != nil {
			for k, item := range componentCfg.GetCustomParams() {
				customParams[k] = &configcenter.ParamItem{Value: item.GetValue(), Comment: item.GetComment()}
			}
		}

		// If customParams is empty, set ClearCustomParams to true to clear all existing custom params.
		clearCustomParams := len(customParams) == 0

		componentConfigurationReq.Requests = append(
			componentConfigurationReq.Requests,
			&configcenter.UpdateClusterConfigurationFileRequest{
				ClusterType:           configcenter.ClusterType_DEDICATED_TIDB,
				ClusterId:             strconv.FormatUint(cluster.ClusterId, 10),
				ConfigurationFilename: componentCfg.ComponentType.String(),
				CustomParams:          customParams,
				ClearCustomParams:     clearCustomParams,
			},
		)
	}

	// update image tag
	if len(componentImageTagReq.Requests) > 0 {
		_, err := cc.configCenterCli.BatchUpdateClusterComponents(ctx, componentImageTagReq)
		if err != nil {
			s, ok := status.FromError(err)
			if ok {
				return bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
			}
			return bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", err.Error())
		}
	}

	// update configuration
	if len(componentConfigurationReq.Requests) > 0 {
		_, err := cc.configCenterCli.BatchUpdateClusterConfigurationFiles(ctx, componentConfigurationReq)
		if err != nil {
			s, ok := status.FromError(err)
			if ok {
				return bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
			}
			return bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", err.Error())
		}
	}

	return nil
}

func (cc *configClientImpl) UpdateClusterMeta(ctx context.Context, req *configcenter.AutoUpdateClusterMetaRequest) bizerr.BizError {
	_, err := cc.configCenterCli.AutoUpdateClusterMeta(ctx, req)
	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			return bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
		}
		return bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", err.Error())
	}
	return nil
}

func (cc *configClientImpl) DeleteClusterMeta(ctx context.Context, cluster *entity.Cluster) bizerr.BizError {
	_, err := cc.configCenterCli.DeleteClusterMeta(ctx, &configcenter.DeleteClusterMetaRequest{
		ClusterType: configcenter.ClusterType_DEDICATED_TIDB,
		ClusterId:   strconv.FormatUint(cluster.ClusterId, 10),
	})
	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			return bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", s.Message())
		}
		return bizerr.New(bizerr.ErrCodeConfigCenterAPI, "%s", err.Error())
	}
	return nil
}