package clusterconfig

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

	"github.com/gogo/protobuf/proto"
	"github.com/samber/lo"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	configcenter "github.com/tidbcloud/config-center/proto/config_center"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

const (
	InitSql       = "INIT_SQL"
	TiDBSemConfig = "tidb_sem"
	// TiDBDefaultConfigFileName is the config file name for default tidb group config
	// It is derived from sharedpool tidb template and used as template for tidb-{group} configs
	TiDBDefaultConfigFileName = "tidb-default"
)

// ParseTenantId parses a tenant ID string into a uint64 pointer
func ParseTenantId(tenantId string, allowMultiTenant bool) (*uint64, error) {
	var t *uint64
	// Empty or multi-tenant tag is valid - return nil pointer with no error
	if allowMultiTenant && (tenantId == "" || tenantId == schemav1.MultiTenantTag) {
		return t, nil
	}
	tenantIdUint, err := strconv.ParseUint(tenantId, 10, 64)
	if err != nil {
		return nil, fmt.Errorf("parse tenantId %s: %w", tenantId, err)
	}

	t = &tenantIdUint
	return t, nil
}

// ParseMetaProfileID parses the optional Config Center meta profile ID.
// An empty ID means that no profile is configured.
func ParseMetaProfileID(id string) (*int64, error) {
	var mid *int64
	if id == "" {
		return mid, nil
	}
	idInt, err := strconv.ParseInt(id, 10, 64)
	if err != nil {
		return nil, fmt.Errorf("parse meta profile id %s: %w", id, err)
	}
	mid = &idInt
	return mid, nil
}

func (c *configClientImpl) GetClient() configcenter.ConfigCenterServiceClient {
	return c.configCenterCli
}

func (c *configClientImpl) GetDefaultVersion(ctx context.Context, tenantId string) (string, error) {
	tenantIdUint, err := ParseTenantId(tenantId, true)
	if err != nil {
		return "", err
	}

	resp, err := c.configCenterCli.GetDefaultVersion(ctx, &configcenter.GetDefaultVersionRequest{
		ClusterType: configcenter.ClusterType_SHARED_POOL,
		TenantId:    tenantIdUint,
		// MetaProfileId: nil,
	})
	if err != nil {
		return "", fmt.Errorf("get cluster default version: %s", err.Error())
	}

	if resp.Version == "" {
		return "", fmt.Errorf("cluster default version is empty")
	}

	return resp.Version, nil
}

// GetEffectiveComponentImageTag resolves the image tag that Config Center will
// use when it initializes a new SharedPool component. A meta profile may
// override both the component version and its image tag, so both overrides
// must be applied before creation decides which component topology to persist.
func (c *configClientImpl) GetEffectiveComponentImageTag(ctx context.Context, sharedPool *entity.SharedPool, componentType entity.ComponentType) (string, error) {
	if sharedPool == nil {
		return "", fmt.Errorf("shared pool is nil")
	}
	if sharedPool.GetVersion() == "" {
		return "", fmt.Errorf("shared pool version is empty")
	}

	tenantID, err := ParseTenantId(sharedPool.TenantId, true)
	if err != nil {
		return "", err
	}
	metaProfileID, err := ParseMetaProfileID(sharedPool.ConfigCenterMetaProfileId)
	if err != nil {
		return "", err
	}

	componentName := getConfigCenterComponentName(componentType)
	componentVersion := sharedPool.GetVersion()
	var metaProfile *configcenter.MetaProfile
	if metaProfileID != nil {
		metaProfile, err = c.configCenterCli.GetMetaProfile(ctx, &configcenter.GetMetaProfileRequest{
			ClusterType: configcenter.ClusterType_SHARED_POOL,
			ProfileId:   *metaProfileID,
		})
		if err != nil {
			return "", fmt.Errorf("get meta profile %d: %w", *metaProfileID, err)
		}
		if version, ok := metaProfile.GetComponentVersions()[componentName]; ok {
			componentVersion = version
		}
	}

	image, err := c.configCenterCli.GetComponentImage(ctx, &configcenter.GetComponentImageRequest{
		ClusterType:      configcenter.ClusterType_SHARED_POOL,
		ComponentName:    componentName,
		ComponentVersion: componentVersion,
		TenantId:         tenantID,
	})
	if err != nil {
		return "", fmt.Errorf("get effective image for component %s: %w", componentName, err)
	}

	imageTag := image.GetTag()
	if metaProfile != nil {
		if overwriteTag := metaProfile.GetComponentImageTags()[componentName]; overwriteTag != "" {
			imageTag = overwriteTag
		}
	}
	if strings.TrimSpace(imageTag) == "" {
		return "", fmt.Errorf("effective image tag is empty for component %s", componentName)
	}

	return imageTag, nil
}

// getTiProxyDefaultVersion gets the default version for tiproxy component
// TiProxy has independent versioning separate from other components
func (c *configClientImpl) getTiProxyDefaultVersion(ctx context.Context, tenantId *uint64) (string, error) {
	resp, err := c.configCenterCli.GetDefaultVersion(ctx, &configcenter.GetDefaultVersionRequest{
		ClusterType: configcenter.ClusterType_SHARED_POOL,
		Component:   strings.ToLower(apiv1.ComponentType_TIPROXY.String()),
		TenantId:    tenantId,
	})
	if err != nil {
		return "", fmt.Errorf("get tiproxy default version: %w", err)
	}

	if resp.Version == "" {
		return "", fmt.Errorf("tiproxy default version is empty")
	}

	return resp.Version, nil
}

func (c *configClientImpl) getDMDefaultVersion(ctx context.Context, tenantId *uint64) (string, error) {
	resp, err := c.configCenterCli.GetDefaultVersion(ctx, &configcenter.GetDefaultVersionRequest{
		ClusterType: configcenter.ClusterType_SHARED_POOL,
		Component:   getConfigCenterComponentName(apiv1.ComponentType_DATA_MIGRATION),
		TenantId:    tenantId,
	})
	if err != nil {
		return "", fmt.Errorf("get dm default version: %w", err)
	}

	if resp.Version == "" {
		return "", fmt.Errorf("dm default version is empty")
	}

	return resp.Version, nil
}

func getDMComponentName() string {
	return getConfigCenterComponentName(apiv1.ComponentType_DATA_MIGRATION)
}

func (c *configClientImpl) CreateClusterMeta(ctx context.Context, sharedPool *entity.SharedPool, tenantId, metaProfileId string) error {
	// sharedpool.version must not be empty
	if sharedPool.GetVersion() == "" {
		return fmt.Errorf("sharedpool version is empty, cannot create cluster meta")
	}

	// first check if the cluster meta already exists
	_, err := c.configCenterCli.GetClusterMeta(ctx, &configcenter.GetClusterMetaRequest{
		ClusterType: configcenter.ClusterType_SHARED_POOL,
		ClusterId:   sharedPool.GetId(),
	})

	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			if s.Code() != codes.NotFound {
				return fmt.Errorf("get cluster meta: %s", s.Message())
			}
		}
	} else {
		tilog.Info(ctx, "shared pool meta already exists, skip creating", zap.String("sharedPoolId", sharedPool.GetId()))
		return nil
	}

	tenantIdUint, err := ParseTenantId(tenantId, true)
	if err != nil {
		return err
	}

	metaProfileIdUint, err := ParseMetaProfileID(metaProfileId)
	if err != nil {
		return err
	}

	metaReq := &configcenter.InitClusterMetaRequest{
		ClusterType:   configcenter.ClusterType_SHARED_POOL,
		ClusterId:     sharedPool.GetId(),
		MetaProfileId: metaProfileIdUint,
		Components: make(
			[]*configcenter.CreateClusterComponentRequest,
			0,
			len(sharedPool.GetComponentGroups())+2, // include tidb and tiproxy
		),
		ConfigurationFiles: make(
			[]*configcenter.CreateClusterConfigFileRequest,
			0,
			len(sharedPool.GetComponentGroups())+2, // include tidb and tiproxy
		),
	}

	// full component image and configuration
	for _, component := range sharedPool.GetComponentGroups() {

		if component.GetComponentType() == apiv1.ComponentType_TIPROXY || component.GetComponentType() == apiv1.ComponentType_TIDB {
			continue
		}

		metaReq.ConfigurationFiles = append(metaReq.ConfigurationFiles,
			&configcenter.CreateClusterConfigFileRequest{
				ClusterType:           configcenter.ClusterType_SHARED_POOL,
				ClusterId:             sharedPool.GetId(),
				ConfigurationFilename: getClusterConfigName(component.GetComponentType(), component.GetId()),
				ConfigurationType:     getConfigCenterComponentName(component.GetComponentType()),
				BelongToComponent:     proto.String(getConfigCenterComponentName(component.GetComponentType())),
				Format:                configcenter.Format_TOML,
				SystemTemplateSource:  buildSystemSource(sharedPool.GetVersion(), tenantIdUint),
				CustomParamsSource:    BuildCustomSourceFromTenant(tenantIdUint),
			})

		if component.GetComponentType() == apiv1.ComponentType_TIFLASH_LEARNER || component.GetComponentType() == apiv1.ComponentType_TIFLASH_COMPUTE_LEARNER {
			continue
		}
		metaReq.Components = append(metaReq.Components, &configcenter.CreateClusterComponentRequest{
			ClusterType:       configcenter.ClusterType_SHARED_POOL,
			ClusterId:         sharedPool.GetId(),
			ComponentName:     getConfigCenterComponentName(component.GetComponentType()),
			OverwriteImageTag: proto.String(""),
			Version:           sharedPool.GetVersion(),
			TenantId:          tenantIdUint,
		})
	}

	// Get default tiproxy version - tiproxy has its own versioning separate from TiDB
	tiproxyVersion, err := c.getTiProxyDefaultVersion(ctx, tenantIdUint)
	if err != nil {
		// Fallback to TiDB version if tiproxy version not available
		tilog.Warn(ctx, "failed to get tiproxy default version, fallback to tidb version",
			zap.String("sharedPoolId", sharedPool.GetId()),
			zap.Error(err))
		tiproxyVersion = sharedPool.GetVersion()
	}
	// Add tidb component and configuration file template for sharedpool
	// These are used as templates for logical clusters to derive from
	tilog.Info(ctx, "creating tidb component and config template for sharedpool",
		zap.String("sharedPoolId", sharedPool.GetId()),
		zap.String("tidbVersion", sharedPool.GetVersion()),
		zap.String("tiproxyVersion", tiproxyVersion))

	metaReq.Components = append(metaReq.Components, &configcenter.CreateClusterComponentRequest{
		ClusterType:       configcenter.ClusterType_SHARED_POOL,
		ClusterId:         sharedPool.GetId(),
		ComponentName:     strings.ToLower(apiv1.ComponentType_TIDB.String()),
		OverwriteImageTag: proto.String(""),
		Version:           sharedPool.GetVersion(),
		TenantId:          tenantIdUint,
	})
	metaReq.ConfigurationFiles = append(metaReq.ConfigurationFiles,
		&configcenter.CreateClusterConfigFileRequest{
			ClusterType:           configcenter.ClusterType_SHARED_POOL,
			ClusterId:             sharedPool.GetId(),
			ConfigurationFilename: strings.ToLower(apiv1.ComponentType_TIDB.String()),
			ConfigurationType:     strings.ToLower(apiv1.ComponentType_TIDB.String()),
			BelongToComponent:     proto.String(strings.ToLower(apiv1.ComponentType_TIDB.String())),
			Format:                configcenter.Format_TOML,
			SystemTemplateSource:  buildSystemSource(sharedPool.GetVersion(), tenantIdUint),
			CustomParamsSource:    BuildCustomSourceFromTenant(tenantIdUint),
		})

	// Add tiproxy component and configuration file template for sharedpool
	// Note: tiproxy uses its own version, not the TiDB version
	metaReq.Components = append(metaReq.Components, &configcenter.CreateClusterComponentRequest{
		ClusterType:       configcenter.ClusterType_SHARED_POOL,
		ClusterId:         sharedPool.GetId(),
		ComponentName:     strings.ToLower(apiv1.ComponentType_TIPROXY.String()),
		OverwriteImageTag: proto.String(""),
		Version:           tiproxyVersion,
		TenantId:          tenantIdUint,
	})
	metaReq.ConfigurationFiles = append(metaReq.ConfigurationFiles,
		&configcenter.CreateClusterConfigFileRequest{
			ClusterType:           configcenter.ClusterType_SHARED_POOL,
			ClusterId:             sharedPool.GetId(),
			ConfigurationFilename: strings.ToLower(apiv1.ComponentType_TIPROXY.String()),
			ConfigurationType:     strings.ToLower(apiv1.ComponentType_TIPROXY.String()),
			BelongToComponent:     proto.String(strings.ToLower(apiv1.ComponentType_TIPROXY.String())),
			Format:                configcenter.Format_TOML,
			SystemTemplateSource:  buildSystemSource(tiproxyVersion, tenantIdUint),
			CustomParamsSource:    BuildCustomSourceFromTenant(tenantIdUint),
		})

	// init sql is cluster level
	//  tidb sem config is cluster level
	metaReq.ConfigurationFiles = append(metaReq.ConfigurationFiles,
		&configcenter.CreateClusterConfigFileRequest{
			ClusterType:           configcenter.ClusterType_SHARED_POOL,
			ClusterId:             sharedPool.GetId(),
			ConfigurationFilename: InitSql, // "INIT_SQL"
			ConfigurationType:     InitSql,
			Format:                configcenter.Format_LINES,
			SystemTemplateSource:  buildSystemSource(sharedPool.GetVersion(), tenantIdUint),
			CustomParamsSource:    BuildCustomSourceFromTenant(tenantIdUint),
		},
		&configcenter.CreateClusterConfigFileRequest{
			ClusterType:           configcenter.ClusterType_SHARED_POOL,
			ClusterId:             sharedPool.GetId(),
			ConfigurationFilename: TiDBSemConfig,
			ConfigurationType:     TiDBSemConfig,
			Format:                configcenter.Format_JSON,
			SystemTemplateSource:  buildSystemSource(sharedPool.GetVersion(), tenantIdUint),
			CustomParamsSource:    BuildCustomSourceFromTenant(tenantIdUint),
		})

	_, err = c.configCenterCli.InitClusterMeta(ctx, metaReq)
	if err != nil {
		return fmt.Errorf("create cluster meta: %w", err)
	}

	tilog.Info(ctx, "created sharedpool meta with tidb/tiproxy templates",
		zap.String("sharedPoolId", sharedPool.GetId()),
		zap.Int("components", len(metaReq.Components)),
		zap.Int("configFiles", len(metaReq.ConfigurationFiles)))

	return nil
}

func (c *configClientImpl) CreateNewComponentConfig(ctx context.Context, sharedPool *entity.SharedPool, componentGroup *entity.Component) error {
	// now only support tikv group config creation
	if componentGroup.GetComponentType() != apiv1.ComponentType_TIKV {
		return fmt.Errorf("create component config not support for %s", componentGroup.GetComponentType().String())
	}

	// first check if the tikv group config already exists
	_, err := c.configCenterCli.GetClusterConfigurationFile(ctx, &configcenter.GetClusterConfigurationFileRequest{
		ClusterType:           configcenter.ClusterType_SHARED_POOL,
		ClusterId:             sharedPool.GetId(),
		ConfigurationFilename: getClusterConfigName(apiv1.ComponentType_TIKV, componentGroup.GetId()),
	})

	if err != nil {
		s, ok := status.FromError(err)
		if !ok {
			return fmt.Errorf("get tikv group config: %w", err)
		}
		if s.Code() != codes.NotFound {
			return fmt.Errorf("get tikv group config: %s", s.Message())
		}
	} else {
		tilog.Info(ctx, "tikv group config already exists, skip creating", zap.String("componentGroupId", componentGroup.GetId()))
		return nil
	}

	// get default tikv config from shared pool
	// default tikv component id is same as PD id (hash of shared pool id)
	defaultTikvConfig, err := c.getClusterConfigurationFile(ctx, sharedPool, &entity.Component{
		Id:            sharedPool.GetDefaultComponenId(),
		ComponentType: apiv1.ComponentType_TIKV,
	})
	if err != nil {
		return fmt.Errorf("get config from default tikv group: %w", err)
	}

	// create new config group config
	_, err = c.configCenterCli.CreateClusterConfigurationFile(ctx, &configcenter.CreateClusterConfigurationFileRequest{
		ClusterType: configcenter.ClusterType_SHARED_POOL,
		ClusterId:   sharedPool.GetId(),
		ConfigurationFile: &configcenter.ConfigurationFile{
			ClusterType:           configcenter.ClusterType_SHARED_POOL,
			ConfigurationFilename: getClusterConfigName(apiv1.ComponentType_TIKV, componentGroup.GetId()),
			ConfigurationType:     strings.ToLower(apiv1.ComponentType_TIKV.String()),
			BelongToComponent:     strings.ToLower(apiv1.ComponentType_TIKV.String()),
			Format:                configcenter.Format_TOML,
			TemplateId:            defaultTikvConfig.GetTemplateId(),
			CustomParams:          defaultTikvConfig.GetCustomParams(),
		},
	})
	if err != nil {
		return fmt.Errorf("create tikv group config: %s", err.Error())
	}

	return nil
}

func (c *configClientImpl) DeleteTiKVComponentConfig(ctx context.Context, sharedPool *entity.SharedPool, componentGroup *entity.Component) error {
	if componentGroup.GetComponentType() != apiv1.ComponentType_TIKV {
		return fmt.Errorf("delete tikv component config not support for %s", componentGroup.GetComponentType().String())
	}

	_, err := c.configCenterCli.DeleteClusterConfigurationFile(ctx, &configcenter.DeleteClusterConfigurationFileRequest{
		ClusterType:           configcenter.ClusterType_SHARED_POOL,
		ClusterId:             sharedPool.GetId(),
		ConfigurationFilename: getClusterConfigName(apiv1.ComponentType_TIKV, componentGroup.GetId()),
	})
	if err != nil {
		s, ok := status.FromError(err)
		if ok && s.Code() == codes.NotFound {
			tilog.Info(ctx, "tikv group config already deleted, skip",
				zap.String("sharedPoolId", sharedPool.GetId()),
				zap.String("componentGroupId", componentGroup.GetId()))
			return nil
		}
		return fmt.Errorf("delete tikv group config: %w", err)
	}

	return nil
}

func getClusterConfigName(componentType apiv1.ComponentType, componentGroupId string) string {
	configName := strings.ToLower(componentType.String())
	if componentType == apiv1.ComponentType_TIKV {
		configName = fmt.Sprintf("%s-%s", configName, componentGroupId)
	}
	return configName
}

func getConfigCenterComponentName(componentType apiv1.ComponentType) string {
	return strings.ToLower(componentType.String())
}

func buildSystemSource(version string, tenantIdUint *uint64) *configcenter.CreateClusterConfigFileRequest_FromSku {
	source := &configcenter.CreateClusterConfigFileRequest_FromSku{FromSku: &configcenter.FromSKU{
		Version: version,
	}}

	if tenantIdUint != nil {
		source.FromSku.TenantId = lo.FromPtr(tenantIdUint)
	}
	return source
}

func BuildCustomSourceFromTenant(tenantIdUint *uint64) *configcenter.CreateClusterConfigFileRequest_FromTenant {
	if tenantIdUint != nil {
		return &configcenter.CreateClusterConfigFileRequest_FromTenant{FromTenant: &configcenter.FromTenant{
			TenantId: lo.FromPtr(tenantIdUint),
		}}
	}
	return nil
}

var (
	AllBRComponents = map[string]bool{
		strings.ToLower(apiv1.ComponentType_TIBR_AUTOBACKUP.String()):   true,
		strings.ToLower(apiv1.ComponentType_TIBR_RESTORE.String()):      true,
		strings.ToLower(apiv1.ComponentType_TIBRGC_T2_DFSGC.String()):   true,
		strings.ToLower(apiv1.ComponentType_TIBRGC_T3_ARCHIVE.String()): true,
		strings.ToLower(apiv1.ComponentType_CLASSIC_RESTORE.String()):   true,
	}
	AllExportComponents = map[string]bool{
		strings.ToLower(apiv1.ComponentType_DUMPLING.String()): true,
	}
)

func (c *configClientImpl) AddBRComponentsToClusterMeta(ctx context.Context, sharedPool *entity.SharedPool, tenantId, metaProfileId string) error {
	clusterMeta, err := c.configCenterCli.GetClusterMeta(ctx, &configcenter.GetClusterMetaRequest{
		ClusterType: configcenter.ClusterType_SHARED_POOL,
		ClusterId:   sharedPool.GetId(),
	})
	if err != nil {
		tilog.Error(ctx, "get cluster meta", zap.Error(err))
		return fmt.Errorf("get cluster meta: %s", err.Error())
	}

	noExistBRComponents := getNoExistBRComponents(clusterMeta.GetComponents())
	noExistBRConfigFiles := getNoExistBRConfigFiles(clusterMeta.GetConfigurationFiles())
	if len(noExistBRComponents) == 0 && len(noExistBRConfigFiles) == 0 {
		tilog.Info(ctx, "all br components already exist, skip adding", zap.String("sharedPoolId", sharedPool.GetId()))
		return nil
	}
	tilog.Info(
		ctx,
		"adding br components to cluster meta",
		zap.String("sharedPoolId", sharedPool.GetId()),
		zap.Strings("noExistBRComponents", noExistBRComponents),
		zap.Strings("noExistBRConfigFiles", noExistBRConfigFiles),
	)

	tenantIdUint, err := ParseTenantId(tenantId, true)
	if err != nil {
		return err
	}

	metaProfileIdUint, err := ParseMetaProfileID(metaProfileId)
	if err != nil {
		return err
	}

	version := sharedPool.GetVersion()
	metaReq := &configcenter.InitClusterMetaRequest{
		ClusterType:        configcenter.ClusterType_SHARED_POOL,
		ClusterId:          sharedPool.GetId(),
		MetaProfileId:      metaProfileIdUint,
		Components:         make([]*configcenter.CreateClusterComponentRequest, 0, len(noExistBRComponents)),
		ConfigurationFiles: make([]*configcenter.CreateClusterConfigFileRequest, 0, len(noExistBRConfigFiles)),
	}

	for _, componentName := range noExistBRComponents {
		metaReq.Components = append(metaReq.Components, &configcenter.CreateClusterComponentRequest{
			ClusterType:       configcenter.ClusterType_SHARED_POOL,
			ClusterId:         sharedPool.GetId(),
			ComponentName:     componentName,
			OverwriteImageTag: proto.String(""),
			Version:           version,
			TenantId:          tenantIdUint,
		})
	}

	for _, componentName := range noExistBRConfigFiles {
		metaReq.ConfigurationFiles = append(metaReq.ConfigurationFiles, &configcenter.CreateClusterConfigFileRequest{
			ClusterType:           configcenter.ClusterType_SHARED_POOL,
			ClusterId:             sharedPool.GetId(),
			ConfigurationFilename: componentName,
			ConfigurationType:     componentName,
			BelongToComponent:     proto.String(componentName),
			Format:                configcenter.Format_TOML,
			SystemTemplateSource:  buildSystemSource(version, tenantIdUint),
			CustomParamsSource:    BuildCustomSourceFromTenant(tenantIdUint),
		})
	}

	_, err = c.configCenterCli.InitClusterMeta(ctx, metaReq)
	if err != nil {
		tilog.Error(ctx, "add br components to cluster meta", zap.Error(err))
		return fmt.Errorf("add br components to cluster meta: %s", err.Error())
	}
	tilog.Info(ctx, "added br components to cluster meta", zap.String("sharedPoolId", sharedPool.GetId()), zap.Strings("noExistBRComponents", noExistBRComponents))
	return nil
}

// EnsurePDMicroserviceComponents reconciles missing component metadata and configuration files for a PD microservice topology.
func (c *configClientImpl) EnsurePDMicroserviceComponents(ctx context.Context, sharedPool *entity.SharedPool) error {
	if sharedPool == nil {
		return fmt.Errorf("shared pool is nil")
	}

	clusterMeta, err := c.configCenterCli.GetClusterMeta(ctx, &configcenter.GetClusterMetaRequest{
		ClusterType: configcenter.ClusterType_SHARED_POOL,
		ClusterId:   sharedPool.GetId(),
	})
	if err != nil {
		return fmt.Errorf("get cluster meta: %w", err)
	}

	targets := make(map[string]struct{})
	for _, component := range sharedPool.GetComponentGroups() {
		if component == nil {
			continue
		}
		switch component.GetComponentType() {
		case apiv1.ComponentType_PD_TSO, apiv1.ComponentType_PD_SCHEDULING, apiv1.ComponentType_RESOURCE_MANAGER:
			targets[getConfigCenterComponentName(component.GetComponentType())] = struct{}{}
		}
	}

	existingComponentVersions := make(map[string]string, len(clusterMeta.GetComponents()))
	for _, component := range clusterMeta.GetComponents() {
		existingComponentVersions[strings.ToLower(component.GetComponentName())] = component.GetVersion()
	}
	existingConfigs := make(map[string]struct{}, len(clusterMeta.GetConfigurationFiles()))
	for _, configuration := range clusterMeta.GetConfigurationFiles() {
		existingConfigs[strings.ToLower(configuration.GetConfigurationFilename())] = struct{}{}
	}

	targetNames := make([]string, 0, len(targets))
	for componentName := range targets {
		targetNames = append(targetNames, componentName)
	}
	sort.Strings(targetNames)

	missingComponents := make([]string, 0, len(targetNames))
	missingConfigs := make([]string, 0, len(targetNames))
	for _, componentName := range targetNames {
		if _, ok := existingComponentVersions[componentName]; !ok {
			missingComponents = append(missingComponents, componentName)
		}
		if _, ok := existingConfigs[componentName]; !ok {
			missingConfigs = append(missingConfigs, componentName)
		}
	}
	// Validate tenant and profile metadata even when reconciliation is a no-op.
	// The apply action continues to render Dataplane resources after this method
	// returns, so the fast path must preserve the same fail-closed guarantees as
	// the initialization path.
	tenantIDString, err := entity.ResolveSharedPoolTenantID(sharedPool)
	if err != nil {
		return fmt.Errorf("resolve shared pool tenant id: %w", err)
	}
	tenantID, err := ParseTenantId(tenantIDString, true)
	if err != nil {
		return fmt.Errorf("parse shared pool tenant id: %w", err)
	}
	metaProfileID, err := ParseMetaProfileID(sharedPool.ConfigCenterMetaProfileId)
	if err != nil {
		return err
	}
	if len(missingComponents) == 0 && len(missingConfigs) == 0 {
		return nil
	}
	tenantSource := "dataplane-tag"
	if sharedPool.TenantId != "" {
		tenantSource = "entity"
	} else if sharedPool.GetProvider().GetTags()[schemav1.LabelUserTenantKey] == "" {
		tenantSource = "unavailable"
	}
	tilog.Info(ctx, "ensuring pd microservice components in config center",
		zap.String("sharedPoolId", sharedPool.GetId()),
		zap.String("dataplaneId", sharedPool.GetDataplaneName()),
		zap.String("tenantSource", tenantSource),
		zap.Bool("hasMetaProfile", sharedPool.HasConfigCenterMetaProfile()),
		zap.Strings("missingComponents", missingComponents),
		zap.Strings("missingConfigs", missingConfigs))

	version := sharedPool.GetVersion()
	if version == "" {
		version = clusterMeta.GetVersion()
	}
	if version == "" {
		for _, component := range clusterMeta.GetComponents() {
			if strings.EqualFold(component.GetComponentName(), getConfigCenterComponentName(apiv1.ComponentType_PD)) {
				version = component.GetVersion()
				break
			}
		}
	}
	requiresFallbackVersion := len(missingComponents) > 0
	for _, componentName := range missingConfigs {
		if existingComponentVersions[componentName] == "" {
			requiresFallbackVersion = true
			break
		}
	}
	if version == "" && requiresFallbackVersion {
		return fmt.Errorf("sharedpool and cluster meta versions are empty, cannot ensure pd microservice components")
	}

	request := &configcenter.InitClusterMetaRequest{
		ClusterType:        configcenter.ClusterType_SHARED_POOL,
		ClusterId:          sharedPool.GetId(),
		MetaProfileId:      metaProfileID,
		Components:         make([]*configcenter.CreateClusterComponentRequest, 0, len(missingComponents)),
		ConfigurationFiles: make([]*configcenter.CreateClusterConfigFileRequest, 0, len(missingConfigs)),
	}
	for _, componentName := range missingComponents {
		request.Components = append(request.Components, &configcenter.CreateClusterComponentRequest{
			ClusterType:       configcenter.ClusterType_SHARED_POOL,
			ClusterId:         sharedPool.GetId(),
			ComponentName:     componentName,
			OverwriteImageTag: proto.String(""),
			Version:           version,
			TenantId:          tenantID,
		})
	}
	for _, componentName := range missingConfigs {
		configVersion := existingComponentVersions[componentName]
		if configVersion == "" {
			configVersion = version
		}
		configurationRequest := &configcenter.CreateClusterConfigFileRequest{
			ClusterType:           configcenter.ClusterType_SHARED_POOL,
			ClusterId:             sharedPool.GetId(),
			ConfigurationFilename: componentName,
			ConfigurationType:     componentName,
			BelongToComponent:     proto.String(componentName),
			Format:                configcenter.Format_TOML,
			SystemTemplateSource:  buildSystemSource(configVersion, tenantID),
		}
		// A typed nil assigned to a protobuf oneof interface is non-nil and can
		// panic during access or marshaling. Multi-tenant pools intentionally
		// have no tenant-specific custom parameter source, so leave the oneof
		// completely unset in that case.
		if tenantID != nil {
			configurationRequest.CustomParamsSource = BuildCustomSourceFromTenant(tenantID)
		}
		request.ConfigurationFiles = append(request.ConfigurationFiles, configurationRequest)
	}

	if _, err := c.configCenterCli.InitClusterMeta(ctx, request); err != nil {
		return fmt.Errorf("ensure pd microservice components: %w", err)
	}
	return nil
}

func getNoExistBRComponents(components []*configcenter.ClusterComponent) []string {
	existBRComponents := map[string]bool{}
	for _, component := range components {
		componentName := strings.ToLower(component.ComponentName)
		if _, ok := AllBRComponents[componentName]; ok {
			existBRComponents[componentName] = true
		}
	}

	nonExistBRComponents := []string{}
	for componentName := range AllBRComponents {
		if _, ok := existBRComponents[componentName]; !ok {
			nonExistBRComponents = append(nonExistBRComponents, componentName)
		}
	}
	return nonExistBRComponents
}

func getNoExistBRConfigFiles(configFiles []*configcenter.ConfigurationFile) []string {
	existBRConfigFiles := map[string]bool{}
	for _, configFile := range configFiles {
		configName := strings.ToLower(configFile.GetConfigurationFilename())
		if _, ok := AllBRComponents[configName]; ok {
			existBRConfigFiles[configName] = true
		}
	}

	nonExistBRConfigFiles := []string{}
	for componentName := range AllBRComponents {
		if _, ok := existBRConfigFiles[componentName]; !ok {
			nonExistBRConfigFiles = append(nonExistBRConfigFiles, componentName)
		}
	}
	return nonExistBRConfigFiles
}

func (c *configClientImpl) AddExportComponentsToClusterMeta(ctx context.Context, tidbCluster *entity.TiDBCluster) error {
	logicalClusterId := tidbCluster.GetId()
	clusterMeta, err := c.configCenterCli.GetClusterMeta(ctx, &configcenter.GetClusterMetaRequest{
		ClusterType: configcenter.ClusterType_LOGICAL_CLUSTER,
		ClusterId:   logicalClusterId,
	})
	if err != nil {
		tilog.Error(ctx, "get export logical cluster meta", zap.Error(err), zap.String("logicalClusterId", logicalClusterId))
		return fmt.Errorf("get export logical cluster meta: %w", err)
	}

	noExistExportComponents := getNoExistExportComponents(clusterMeta.GetComponents())
	if len(noExistExportComponents) == 0 {
		tilog.Info(ctx, "all export components already exist, skip adding",
			zap.String("logicalClusterId", logicalClusterId),
			zap.String("sharedPoolId", tidbCluster.GetSharedPoolId()))
		return nil
	}

	tilog.Info(ctx, "adding export components to logical cluster meta",
		zap.String("logicalClusterId", logicalClusterId),
		zap.String("sharedPoolId", tidbCluster.GetSharedPoolId()),
		zap.Strings("noExistExportComponents", noExistExportComponents))

	tenantIdUint, err := ParseTenantId(tidbCluster.GetTenantId(), false)
	if err != nil {
		return err
	}

	version := tidbCluster.GetVersion()
	metaReq := &configcenter.AutoUpdateClusterMetaRequest{
		ClusterType: configcenter.ClusterType_LOGICAL_CLUSTER,
		ClusterId:   logicalClusterId,
		Version:     proto.String(version),
		ComponentRequests: make(
			[]*configcenter.UpdateClusterComponentRequest,
			0,
			len(noExistExportComponents),
		),
		ConfigurationFileRequests: make(
			[]*configcenter.AutoUpdateClusterConfigurationFileRequest,
			0,
			len(noExistExportComponents),
		),
	}

	for _, componentName := range noExistExportComponents {
		metaReq.ConfigurationFileRequests = append(metaReq.ConfigurationFileRequests,
			&configcenter.AutoUpdateClusterConfigurationFileRequest{
				ClusterType:           configcenter.ClusterType_LOGICAL_CLUSTER,
				ClusterId:             logicalClusterId,
				ConfigurationFilename: componentName,
				ConfigurationType:     proto.String(componentName),
				BelongToComponent:     proto.String(componentName),
				Format:                configcenter.Format_TOML.Enum(),
				Version:               version,
				TenantId:              tenantIdUint,
			})
		metaReq.ComponentRequests = append(metaReq.ComponentRequests, &configcenter.UpdateClusterComponentRequest{
			ClusterType:       configcenter.ClusterType_LOGICAL_CLUSTER,
			ClusterId:         logicalClusterId,
			ComponentName:     componentName,
			OverwriteImageTag: proto.String(""),
			AllowMissing:      true,
			Version:           lo.ToPtr(version),
			TenantId:          tenantIdUint,
		})
	}

	_, err = c.configCenterCli.AutoUpdateClusterMeta(ctx, metaReq)
	if err != nil {
		tilog.Error(ctx, "add export components to logical cluster meta", zap.Error(err), zap.String("logicalClusterId", logicalClusterId))
		return fmt.Errorf("add export components to logical cluster meta: %w", err)
	}

	tilog.Info(ctx, "added export components to logical cluster meta",
		zap.String("logicalClusterId", logicalClusterId),
		zap.String("sharedPoolId", tidbCluster.GetSharedPoolId()),
		zap.Strings("noExistExportComponents", noExistExportComponents))
	return nil
}

func getNoExistExportComponents(components []*configcenter.ClusterComponent) []string {
	existExportComponents := map[string]bool{}
	for _, component := range components {
		componentName := strings.ToLower(component.ComponentName)
		if _, ok := AllExportComponents[componentName]; ok {
			existExportComponents[componentName] = true
		}
	}

	nonExistExportComponents := []string{}
	for componentName := range AllExportComponents {
		if _, ok := existExportComponents[componentName]; !ok {
			nonExistExportComponents = append(nonExistExportComponents, componentName)
		}
	}
	return nonExistExportComponents
}

func (c *configClientImpl) CreateTiDBClusterMeta(ctx context.Context, sharedPool *entity.SharedPool, tidbCluster *entity.TiDBCluster) error {
	// tidbCluster.version must not be empty
	if tidbCluster.GetVersion() == "" {
		return fmt.Errorf("tidbCluster version is empty, cannot create tidb cluster meta")
	}

	sharedPoolId := tidbCluster.GetSharedPoolId()
	if sharedPoolId == "" {
		return fmt.Errorf("tidbCluster sharedPoolId is empty, cannot create tidb cluster meta")
	}
	if sharedPool == nil {
		return fmt.Errorf("sharedPool is nil, cannot create tidb cluster meta")
	}
	if sharedPool.GetId() != sharedPoolId {
		return fmt.Errorf("sharedPool id %s does not match tidbCluster sharedPoolId %s", sharedPool.GetId(), sharedPoolId)
	}

	// get first
	_, err := c.configCenterCli.GetClusterMeta(ctx, &configcenter.GetClusterMetaRequest{
		ClusterType: configcenter.ClusterType_LOGICAL_CLUSTER,
		ClusterId:   tidbCluster.GetId(),
	})

	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			// not found, use default version
			if s.Code() != codes.NotFound {
				return fmt.Errorf("get tidbcluster meta: %s", s.Message())
			}
		}
	} else {
		tilog.Info(ctx, "tidb cluster config meta already exists, skip creating", zap.String("tidbClusterId", tidbCluster.GetId()))
		return nil
	}

	tenantId, err := ParseTenantId(tidbCluster.GetTenantId(), false)
	if err != nil {
		return fmt.Errorf("parse tenantId: %w", err)
	}

	// Get tidb template from sharedpool
	tidbTemplateConfig, err := c.configCenterCli.GetClusterConfigurationFile(ctx, &configcenter.GetClusterConfigurationFileRequest{
		ClusterType:           configcenter.ClusterType_SHARED_POOL,
		ClusterId:             sharedPoolId,
		ConfigurationFilename: strings.ToLower(apiv1.ComponentType_TIDB.String()),
	})
	if err != nil {
		return fmt.Errorf("get tidb template config from sharedpool %s: %w", sharedPoolId, err)
	}

	// TiDB version comes from tidbCluster (same as sharedpool version)
	tidbVersion := tidbCluster.GetVersion()
	var tiproxyVersion string

	// Get tidb component from sharedpool (for OverwriteImageTag)
	tidbComponent, err := c.configCenterCli.GetClusterComponent(ctx, &configcenter.GetClusterComponentRequest{
		ClusterType:   configcenter.ClusterType_SHARED_POOL,
		ClusterId:     sharedPoolId,
		ComponentName: strings.ToLower(apiv1.ComponentType_TIDB.String()),
	})
	if err != nil {
		return fmt.Errorf("get tidb component from sharedpool %s: %w", sharedPoolId, err)
	}

	// Create components and configuration files together via InitClusterMeta
	metaReq := &configcenter.InitClusterMetaRequest{
		ClusterType: configcenter.ClusterType_LOGICAL_CLUSTER,
		ClusterId:   tidbCluster.GetId(),
		Components: []*configcenter.CreateClusterComponentRequest{
			{
				ComponentName:     strings.ToLower(apiv1.ComponentType_TIDB.String()),
				OverwriteImageTag: proto.String(tidbComponent.GetEffectiveImageTag()),
				Version:           tidbVersion,
				TenantId:          tenantId,
			},
		},
		ConfigurationFiles: []*configcenter.CreateClusterConfigFileRequest{
			// tidb-default config - derived from sharedpool tidb template
			// This will be used as template for tidb-{group} configs
			{
				ClusterType:           configcenter.ClusterType_LOGICAL_CLUSTER,
				ClusterId:             tidbCluster.GetId(),
				ConfigurationFilename: TiDBDefaultConfigFileName,
				ConfigurationType:     strings.ToLower(apiv1.ComponentType_TIDB.String()),
				BelongToComponent:     proto.String(strings.ToLower(apiv1.ComponentType_TIDB.String())),
				Format:                configcenter.Format_TOML,
				SystemTemplateSource: &configcenter.CreateClusterConfigFileRequest_FromTemplate{
					FromTemplate: &configcenter.FromTemplate{
						TemplateId: tidbTemplateConfig.GetTemplateId(),
					},
				},
				CustomParamsSource: &configcenter.CreateClusterConfigFileRequest_FromInputValues{
					FromInputValues: &configcenter.FromInputValues{
						CustomParams: tidbTemplateConfig.GetCustomParams(),
					},
				},
			},
		},
	}

	if !entity.HasSharedTiProxy(sharedPool) {
		tiproxyTemplateConfig, err := c.configCenterCli.GetClusterConfigurationFile(ctx, &configcenter.GetClusterConfigurationFileRequest{
			ClusterType:           configcenter.ClusterType_SHARED_POOL,
			ClusterId:             sharedPoolId,
			ConfigurationFilename: strings.ToLower(apiv1.ComponentType_TIPROXY.String()),
		})
		if err != nil {
			return fmt.Errorf("get tiproxy template config from sharedpool %s: %w", sharedPoolId, err)
		}

		tiproxyComponent, err := c.configCenterCli.GetClusterComponent(ctx, &configcenter.GetClusterComponentRequest{
			ClusterType:   configcenter.ClusterType_SHARED_POOL,
			ClusterId:     sharedPoolId,
			ComponentName: strings.ToLower(apiv1.ComponentType_TIPROXY.String()),
		})
		if err != nil {
			return fmt.Errorf("get tiproxy component from sharedpool %s: %w", sharedPoolId, err)
		}
		tiproxyVersion = tiproxyComponent.GetVersion()

		metaReq.Components = append(metaReq.Components, &configcenter.CreateClusterComponentRequest{
			ComponentName:     strings.ToLower(apiv1.ComponentType_TIPROXY.String()),
			OverwriteImageTag: proto.String(tiproxyComponent.GetEffectiveImageTag()),
			Version:           tiproxyVersion,
			TenantId:          tenantId,
		})
		metaReq.ConfigurationFiles = append(metaReq.ConfigurationFiles, &configcenter.CreateClusterConfigFileRequest{
			ClusterType:           configcenter.ClusterType_LOGICAL_CLUSTER,
			ClusterId:             tidbCluster.GetId(),
			ConfigurationFilename: strings.ToLower(apiv1.ComponentType_TIPROXY.String()),
			ConfigurationType:     strings.ToLower(apiv1.ComponentType_TIPROXY.String()),
			BelongToComponent:     proto.String(strings.ToLower(apiv1.ComponentType_TIPROXY.String())),
			Format:                configcenter.Format_TOML,
			SystemTemplateSource: &configcenter.CreateClusterConfigFileRequest_FromTemplate{
				FromTemplate: &configcenter.FromTemplate{
					TemplateId: tiproxyTemplateConfig.GetTemplateId(),
				},
			},
			CustomParamsSource: &configcenter.CreateClusterConfigFileRequest_FromInputValues{
				FromInputValues: &configcenter.FromInputValues{
					CustomParams: tiproxyTemplateConfig.GetCustomParams(),
				},
			},
		})
	}

	tilog.Info(ctx, "creating tidb cluster meta from sharedpool templates",
		zap.String("tidbClusterId", tidbCluster.GetId()),
		zap.String("sharedPoolId", sharedPoolId),
		zap.String("tidbVersion", tidbVersion),
		zap.String("tiproxyVersion", tiproxyVersion),
		zap.Bool("sharedTiProxy", entity.HasSharedTiProxy(sharedPool)))

	_, err = c.configCenterCli.InitClusterMeta(ctx, metaReq)
	if err != nil {
		return fmt.Errorf("init tidb cluster meta: %w", err)
	}

	tilog.Info(ctx, "created tidb cluster meta with tidb-default config",
		zap.String("tidbClusterId", tidbCluster.GetId()),
		zap.String("sharedPoolId", sharedPoolId))

	return nil
}

func (c *configClientImpl) CreateTiDBGroupConfig(ctx context.Context, tidbGroup *entity.TiDBGroup) error {
	// first check if the tidb group config already exists
	_, err := c.configCenterCli.GetClusterConfigurationFile(ctx, &configcenter.GetClusterConfigurationFileRequest{
		ClusterType:           configcenter.ClusterType_LOGICAL_CLUSTER,
		ClusterId:             tidbGroup.GetClusterId(),
		ConfigurationFilename: GetTiDBGroupConfigFileName(tidbGroup.GetId()),
	})

	if err != nil {
		s, ok := status.FromError(err)
		if ok {
			if s.Code() != codes.NotFound {
				return fmt.Errorf("get tidb group config: %s", s.Message())
			}
		}
	} else {
		tilog.Info(ctx, "tidb group config already exists, skip creating", zap.String("tidbGroupId", tidbGroup.GetId()))
		return nil
	}

	// get tenant id
	_, err = ParseTenantId(tidbGroup.GetTenantId(), false)
	if err != nil {
		return fmt.Errorf("parse tenantId: %w", err)
	}

	// Get tidb-default config from logical cluster to derive tidb-{group} config
	// tidb-default is guaranteed to exist by migration
	tidbDefaultConfig, err := c.configCenterCli.GetClusterConfigurationFile(ctx, &configcenter.GetClusterConfigurationFileRequest{
		ClusterType:           configcenter.ClusterType_LOGICAL_CLUSTER,
		ClusterId:             tidbGroup.GetClusterId(),
		ConfigurationFilename: TiDBDefaultConfigFileName,
	})
	if err != nil {
		return fmt.Errorf("get tidb-default config: %w", err)
	}

	// Create tidb-{group} config derived from tidb-default using TemplateId + CustomParams
	tilog.Info(ctx, "creating tidb group config from tidb-default",
		zap.String("tidbGroupId", tidbGroup.GetId()),
		zap.String("tidbClusterId", tidbGroup.GetClusterId()),
		zap.Int64("templateId", tidbDefaultConfig.GetTemplateId()))

	_, err = c.configCenterCli.CreateClusterConfigurationFile(ctx, &configcenter.CreateClusterConfigurationFileRequest{
		ClusterType: configcenter.ClusterType_LOGICAL_CLUSTER,
		ClusterId:   tidbGroup.GetClusterId(),
		ConfigurationFile: &configcenter.ConfigurationFile{
			ClusterType:           configcenter.ClusterType_LOGICAL_CLUSTER,
			ConfigurationFilename: GetTiDBGroupConfigFileName(tidbGroup.GetId()),
			ConfigurationType:     strings.ToLower(apiv1.ComponentType_TIDB.String()),
			BelongToComponent:     strings.ToLower(apiv1.ComponentType_TIDB.String()),
			Format:                configcenter.Format_TOML,
			TemplateId:            tidbDefaultConfig.GetTemplateId(),
			CustomParams:          tidbDefaultConfig.GetCustomParams(),
		},
	})
	if err != nil {
		// Handle AlreadyExists as success (idempotency)
		s, ok := status.FromError(err)
		if ok && s.Code() == codes.AlreadyExists {
			tilog.Info(ctx, "tidb group config already exists (idempotent)",
				zap.String("tidbGroupId", tidbGroup.GetId()))
			return nil
		}
		return fmt.Errorf("create tidb group config: %s", err.Error())
	}

	tilog.Info(ctx, "created tidb group config from tidb-default",
		zap.String("tidbGroupId", tidbGroup.GetId()))
	return nil
}

func (c *configClientImpl) DeleteTiDBGroupConfig(ctx context.Context, tidbClusterId, tidbGroupId string) error {
	_, err := c.configCenterCli.DeleteClusterConfigurationFile(ctx, &configcenter.DeleteClusterConfigurationFileRequest{
		ClusterType:           configcenter.ClusterType_LOGICAL_CLUSTER,
		ClusterId:             tidbClusterId,
		ConfigurationFilename: GetTiDBGroupConfigFileName(tidbGroupId),
	})
	if err != nil {
		s, ok := status.FromError(err)
		if ok && s.Code() == codes.NotFound {
			tilog.Info(ctx, "tidb group config already deleted", zap.String("tidbGroupId", tidbGroupId))
			return nil
		}
		return fmt.Errorf("delete tidb group config: %w", err)
	}

	tilog.Info(ctx, "deleted tidb group config",
		zap.String("tidbClusterId", tidbClusterId),
		zap.String("tidbGroupId", tidbGroupId))
	return nil
}

func (c *configClientImpl) DeleteTiDBClusterMeta(ctx context.Context, tidbClusterId string) error {
	return c.deleteClusterMeta(ctx, configcenter.ClusterType_LOGICAL_CLUSTER, tidbClusterId)
}

func (c *configClientImpl) DeleteClusterMeta(ctx context.Context, sharedPoolId string) error {
	return c.deleteClusterMeta(ctx, configcenter.ClusterType_SHARED_POOL, sharedPoolId)
}

func (c *configClientImpl) deleteClusterMeta(ctx context.Context, clusterType configcenter.ClusterType, clusterId string) error {
	_, err := c.configCenterCli.DeleteClusterMeta(ctx, &configcenter.DeleteClusterMetaRequest{
		ClusterType: clusterType,
		ClusterId:   clusterId,
	})
	if err != nil {
		s, ok := status.FromError(err)
		if ok && s.Code() == codes.NotFound {
			tilog.Info(ctx, "cluster meta already deleted",
				zap.String("clusterId", clusterId),
				zap.String("clusterType", clusterType.String()))
			return nil
		}
		return fmt.Errorf("delete cluster meta (type=%s id=%s): %w", clusterType.String(), clusterId, err)
	}

	tilog.Info(ctx, "deleted cluster meta",
		zap.String("clusterId", clusterId),
		zap.String("clusterType", clusterType.String()))
	return nil
}

// not tidb group and tiproxy group config and version
func (c *configClientImpl) GetComponentVersionAndConfig(ctx context.Context, sharedPool *entity.SharedPool, component *entity.Component) (string, *entity.ComponentConfiguration, error) {
	if component.GetComponentType() == apiv1.ComponentType_TIDB {
		return "", nil, fmt.Errorf("get component version and config not support for %s", component.GetComponentType().String())
	}

	var imageResp *configcenter.ClusterComponent
	// tiproxy learner not have image info
	if component.GetComponentType() != apiv1.ComponentType_TIFLASH_LEARNER && component.GetComponentType() != apiv1.ComponentType_TIFLASH_COMPUTE_LEARNER {
		var err error
		imageResp, err = c.configCenterCli.GetClusterComponent(ctx, &configcenter.GetClusterComponentRequest{
			ClusterType:   configcenter.ClusterType_SHARED_POOL,
			ClusterId:     sharedPool.GetId(),
			ComponentName: getConfigCenterComponentName(component.GetComponentType()),
		})
		if err != nil {
			return "", nil, fmt.Errorf("get component version: %w", err)
		}

		if imageResp.GetVersion() == "" {
			return "", nil, fmt.Errorf("component version not found for %s", component.GetComponentType().String())
		}

		// check repo and tag
		if imageResp.BaseImage == "" || imageResp.EffectiveImageTag == "" {
			tilog.Warn(
				ctx,
				"component image not found",
				zap.String("componentType", component.GetComponentType().String()),
				zap.String("sharedPoolId", sharedPool.GetId()),
			)
			return "", nil, fmt.Errorf("component image not found for %s", component.GetComponentType().String())
		}
	}

	configResp, err := c.getClusterConfigurationFile(ctx, sharedPool, component)
	if err != nil {
		return "", nil, err
	}

	compConfig := &entity.ComponentConfiguration{
		ConfigurationFileContent: configResp.GetText(),
		ImageRepo:                imageResp.GetBaseImage(),
		ImageTag:                 imageResp.GetEffectiveImageTag(),
	}
	return imageResp.GetVersion(), compConfig, nil
}

func (c *configClientImpl) GetDMComponentVersionAndConfig(ctx context.Context, sharedPool *entity.SharedPool) (string, *entity.ComponentConfiguration, error) {
	if sharedPool == nil {
		return "", nil, fmt.Errorf("shared pool is nil")
	}
	if sharedPool.GetId() == "" {
		return "", nil, fmt.Errorf("shared pool id is empty")
	}

	imageResp, err := c.getDMClusterComponent(ctx, sharedPool.GetId())
	if err != nil {
		s, ok := status.FromError(err)
		if !ok || s.Code() != codes.NotFound {
			return "", nil, fmt.Errorf("get dm component version: %w", err)
		}
		if err := c.addDMComponentToClusterMeta(ctx, sharedPool); err != nil {
			return "", nil, fmt.Errorf("add dm component to cluster meta: %w", err)
		}
		imageResp, err = c.getDMClusterComponent(ctx, sharedPool.GetId())
		if err != nil {
			return "", nil, fmt.Errorf("get dm component version after backfill: %w", err)
		}
	}

	if imageResp.GetVersion() == "" {
		return "", nil, fmt.Errorf("component version not found for %s", getDMComponentName())
	}
	if imageResp.GetBaseImage() == "" || imageResp.GetEffectiveImageTag() == "" {
		return imageResp.GetVersion(), &entity.ComponentConfiguration{
			ImageRepo: imageResp.GetBaseImage(),
			ImageTag:  imageResp.GetEffectiveImageTag(),
		}, fmt.Errorf("component image not found for %s (sharedPoolId=%s)", getDMComponentName(), sharedPool.GetId())
	}

	return imageResp.GetVersion(), &entity.ComponentConfiguration{
		ImageRepo: imageResp.GetBaseImage(),
		ImageTag:  imageResp.GetEffectiveImageTag(),
	}, nil
}

func (c *configClientImpl) getDMClusterComponent(ctx context.Context, sharedPoolId string) (*configcenter.ClusterComponent, error) {
	return c.configCenterCli.GetClusterComponent(ctx, &configcenter.GetClusterComponentRequest{
		ClusterType:   configcenter.ClusterType_SHARED_POOL,
		ClusterId:     sharedPoolId,
		ComponentName: getDMComponentName(),
	})
}

func (c *configClientImpl) addDMComponentToClusterMeta(ctx context.Context, sharedPool *entity.SharedPool) error {
	tenantIdUint, err := ParseTenantId(sharedPool.TenantId, true)
	if err != nil {
		return err
	}
	dmVersion, err := c.getDMDefaultVersion(ctx, tenantIdUint)
	if err != nil {
		return err
	}

	_, err = c.configCenterCli.InitClusterMeta(ctx, &configcenter.InitClusterMetaRequest{
		ClusterType: configcenter.ClusterType_SHARED_POOL,
		ClusterId:   sharedPool.GetId(),
		Components: []*configcenter.CreateClusterComponentRequest{
			{
				ClusterType:       configcenter.ClusterType_SHARED_POOL,
				ClusterId:         sharedPool.GetId(),
				ComponentName:     getDMComponentName(),
				OverwriteImageTag: proto.String(""),
				Version:           dmVersion,
				TenantId:          tenantIdUint,
			},
		},
	})
	if err != nil {
		return fmt.Errorf("init cluster meta for dm component: %w", err)
	}
	return nil
}

func (c *configClientImpl) getClusterConfigurationFile(ctx context.Context, sharedPool *entity.SharedPool, component *entity.Component) (*configcenter.ConfigurationFile, error) {
	configResp, err := c.configCenterCli.GetClusterConfigurationFile(ctx, &configcenter.GetClusterConfigurationFileRequest{
		ClusterType:           configcenter.ClusterType_SHARED_POOL,
		ClusterId:             sharedPool.GetId(),
		ConfigurationFilename: getClusterConfigName(component.GetComponentType(), component.GetId()),
	})
	if err != nil {
		s, ok := status.FromError(err)
		if !ok {
			// Non-gRPC error (e.g., transport failure or context cancellation)
			// Return immediately without retry
			return nil, fmt.Errorf("get component config: %w", err)
		}

		// Only retry for NotFound error on TIKV component (old style config compatibility)
		if s.Code() == codes.NotFound && component.GetComponentType() == apiv1.ComponentType_TIKV {
			tilog.Warn(
				ctx,
				"tikv group config not found, try old style config",
				zap.String("sharedPoolId", sharedPool.GetId()),
			)
			configResp, err = c.configCenterCli.GetClusterConfigurationFile(ctx, &configcenter.GetClusterConfigurationFileRequest{
				ClusterType:           configcenter.ClusterType_SHARED_POOL,
				ClusterId:             sharedPool.GetId(),
				ConfigurationFilename: strings.ToLower(component.GetComponentType().String()),
			})
			if err == nil {
				return configResp, nil
			}
		}

		return nil, fmt.Errorf("get component config: %w", err)
	}
	return configResp, nil
}

func (c *configClientImpl) GetBRComponentVersionAndConfig(ctx context.Context, sharedPool *entity.SharedPool, tenantId string, componentType entity.ComponentType) (string, *entity.ComponentConfiguration, error) {
	isBRComponent := AllBRComponents[strings.ToLower(componentType.String())]
	if !isBRComponent {
		return "", nil, fmt.Errorf("component %s is not a br component", componentType.String())
	}

	// Notice: this will add br components to cluster meta if not exist for compatibility with old sharedpool
	// try to add missing br components to cluster meta, only add if not exist
	err := c.AddBRComponentsToClusterMeta(ctx, sharedPool, tenantId, "")
	if err != nil {
		return "", nil, fmt.Errorf("add br components to cluster meta: %w", err)
	}

	brVersion, brConfig, err := c.GetComponentVersionAndConfig(ctx, sharedPool, &entity.Component{
		Id:            "",
		ComponentType: componentType,
	})
	if err != nil {
		return "", nil, fmt.Errorf("get component version and config: %w", err)
	}
	// if br component image not found, use tikv image
	if brConfig.ImageRepo == "" || brConfig.ImageTag == "" {
		_, tikvConfig, err := c.GetComponentVersionAndConfig(ctx, sharedPool, &entity.Component{
			Id:            sharedPool.GetDefaultComponenId(),
			ComponentType: apiv1.ComponentType_TIKV,
		})
		if err != nil {
			return "", nil, fmt.Errorf("get tikv component version and config: %w", err)
		}
		if tikvConfig.ImageRepo == "" || tikvConfig.ImageTag == "" {
			return "", nil, fmt.Errorf("tikv image not found")
		}
		brConfig.ImageRepo = tikvConfig.ImageRepo
		brConfig.ImageTag = tikvConfig.ImageTag
	}

	return brVersion, brConfig, nil
}

func (c *configClientImpl) GetExportComponentVersionAndConfig(ctx context.Context, tidbCluster *entity.TiDBCluster, componentType entity.ComponentType) (string, *entity.ComponentConfiguration, error) {
	isExportComponent := AllExportComponents[strings.ToLower(componentType.String())]
	if !isExportComponent {
		return "", nil, fmt.Errorf("component %s is not an export component", componentType.String())
	}

	logicalClusterId := tidbCluster.GetId()

	err := c.AddExportComponentsToClusterMeta(ctx, tidbCluster)
	if err != nil {
		return "", nil, fmt.Errorf("add export components to cluster meta: %w", err)
	}

	imageResp, err := c.configCenterCli.GetClusterComponent(ctx, &configcenter.GetClusterComponentRequest{
		ClusterType:   configcenter.ClusterType_LOGICAL_CLUSTER,
		ClusterId:     logicalClusterId,
		ComponentName: strings.ToLower(componentType.String()),
	})
	if err != nil {
		return "", nil, fmt.Errorf("get component version: %s", err.Error())
	}

	if imageResp.GetVersion() == "" {
		return "", nil, fmt.Errorf("component version not found for %s", componentType.String())
	}

	if imageResp.BaseImage == "" || imageResp.EffectiveImageTag == "" {
		tilog.Error(
			ctx,
			"component image not found",
			zap.String("componentType", componentType.String()),
			zap.String("tidbClusterId", logicalClusterId),
		)
		return "", nil, fmt.Errorf("component image not found for %s", componentType.String())
	}

	configResp, err := c.configCenterCli.GetClusterConfigurationFile(ctx, &configcenter.GetClusterConfigurationFileRequest{
		ClusterType:           configcenter.ClusterType_LOGICAL_CLUSTER,
		ClusterId:             logicalClusterId,
		ConfigurationFilename: strings.ToLower(componentType.String()),
	})
	if err != nil {
		return "", nil, fmt.Errorf("get component config: %s", err.Error())
	}

	exportConfig := &entity.ComponentConfiguration{
		ConfigurationFileContent: configResp.GetText(),
		ImageRepo:                imageResp.GetBaseImage(),
		ImageTag:                 imageResp.GetEffectiveImageTag(),
	}
	return imageResp.GetVersion(), exportConfig, nil
}

func (c *configClientImpl) GetClusterInitSqlStmts(ctx context.Context, sharedPoolId string) (string, error) {
	resp, err := c.configCenterCli.GetClusterConfigurationFile(ctx, &configcenter.GetClusterConfigurationFileRequest{
		ClusterType:           configcenter.ClusterType_SHARED_POOL,
		ClusterId:             sharedPoolId,
		ConfigurationFilename: InitSql, // "INIT_SQL"
	})
	if err != nil {
		return "", fmt.Errorf("get cluster init sql stmts: %s", err.Error())
	}

	if resp.GetText() == "" {
		return "", fmt.Errorf("init sql stmts not found for shared pool %s", sharedPoolId)
	}

	return resp.GetText(), nil
}

func (c *configClientImpl) GetClusterTiDBSemConfig(ctx context.Context, sharedPoolId string) (string, error) {
	resp, err := c.configCenterCli.GetClusterConfigurationFile(ctx, &configcenter.GetClusterConfigurationFileRequest{
		ClusterType:           configcenter.ClusterType_SHARED_POOL,
		ClusterId:             sharedPoolId,
		ConfigurationFilename: TiDBSemConfig,
	})
	if err != nil {
		return "", fmt.Errorf("get cluster tidb sem config: %s", err.Error())
	}

	if resp.GetText() == "" {
		return "", fmt.Errorf("tidb sem config is empty for shared pool %s", sharedPoolId)
	}

	return resp.GetText(), nil
}

func (c *configClientImpl) GetTiDBClusterVersion(ctx context.Context, tidbCluster *entity.TiDBCluster) (string, error) {
	if tidbCluster.GetId() == "" {
		return "", fmt.Errorf("tidb cluster id is empty")
	}

	resp, err := c.configCenterCli.GetClusterMeta(ctx, &configcenter.GetClusterMetaRequest{
		ClusterType:   configcenter.ClusterType_LOGICAL_CLUSTER,
		ClusterId:     tidbCluster.GetId(),
		MainComponent: strings.ToLower(apiv1.ComponentType_TIDB.String()),
	})
	if err != nil {
		return "", err
	}

	if resp.GetVersion() == "" {
		return "", fmt.Errorf("tidb cluster version not found")
	}

	return resp.GetVersion(), nil
}

// tidb group and tiproxy group config and version
func (c *configClientImpl) GetTiDBClusterVersionAndConfig(ctx context.Context, tidbGroup *entity.TiDBGroup, componentType entity.ComponentType) (string, *entity.ComponentConfiguration, error) {
	if componentType != apiv1.ComponentType_TIDB && componentType != apiv1.ComponentType_TIPROXY {
		return "", nil, fmt.Errorf("get tidb group component version and config not support for %s", componentType.String())
	}

	imageResp, err := c.configCenterCli.GetClusterComponent(ctx, &configcenter.GetClusterComponentRequest{
		ClusterType:   configcenter.ClusterType_LOGICAL_CLUSTER,
		ClusterId:     tidbGroup.GetClusterId(),
		ComponentName: strings.ToLower(componentType.String()),
	})
	if err != nil {
		return "", nil, fmt.Errorf("get component version: %s", err.Error())
	}

	if imageResp.GetVersion() == "" {
		return "", nil, fmt.Errorf("component version not found for %s", componentType.String())
	}

	// check repo and tag
	if imageResp.BaseImage == "" || imageResp.EffectiveImageTag == "" {
		tilog.Warn(
			ctx,
			"component image not found",
			zap.String("componentType", componentType.String()),
			zap.String("tidbClusterId", tidbGroup.GetClusterId()),
		)
		return "", nil, fmt.Errorf("component image not found for %s", componentType.String())
	}

	// tidb has multiple group config
	// so config name use tidb group id
	configName := componentType.String()
	if componentType == apiv1.ComponentType_TIDB {
		configName = GetTiDBGroupConfigFileName(tidbGroup.GetId())
	}
	configResp, err := c.configCenterCli.GetClusterConfigurationFile(ctx, &configcenter.GetClusterConfigurationFileRequest{
		ClusterType:           configcenter.ClusterType_LOGICAL_CLUSTER,
		ClusterId:             tidbGroup.GetClusterId(),
		ConfigurationFilename: configName,
	})
	if err != nil {
		return "", nil, fmt.Errorf("get component config: %s", err.Error())
	}

	compConfig := &entity.ComponentConfiguration{
		ConfigurationFileContent: configResp.GetText(),
		ImageRepo:                imageResp.GetBaseImage(),
		ImageTag:                 imageResp.GetEffectiveImageTag(),
	}
	return imageResp.GetVersion(), compConfig, nil
}

func GetTiDBGroupConfigFileName(tidbGroupId string) string {
	// tidb group config file name is "tidb-{tidbGroupId}"
	return fmt.Sprintf("tidb-%s", tidbGroupId)
}