package config

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"strconv"

	baseapi "github.com/tidbcloud/devops/api"
	ccent "github.com/tidbcloud/devops/entity/cloudconfig"
	"github.com/tidbcloud/devops/pkg/infra/errs"
	"github.com/tidbcloud/devops/pkg/infra/utils/container"
	"github.com/tidbcloud/devops/pkg/infra/utils/pointer"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/api"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/change/common"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/cvt"

	ccproto "github.com/tidbcloud/config-center/proto/config_center"
)

type actorCreate struct {
	*baseActor
}

func (a *actorCreate) Preview(param api.PreviewConfigurationChangeParam) (*api.PreviewConfigurationChangeResult, error) {
	ctx := a.Context

	createParam := api.CreateConfigurationChangeParam{
		CreateConfigurationChangeBaseParam: param.CreateConfigurationChangeBaseParam,
		// other fields is not used in preview
	}
	builder, err := a.newConfigurationChange(createParam)
	if err != nil {
		return nil, err
	}

	result := &api.PreviewConfigurationChangeResult{
		Change:        builder.Change,
		Configuration: builder.ChangedConfig,
	}
	newChange := result.Change

	clusterType := ccproto.ClusterType_DEDICATED_TIDB
	if newChange.TargetCluster != nil {
		clusterType = cvt.To.ConfigCenter.ClusterType(newChange.TargetCluster.ClusterType)
	} else if newChange.TargetTenantInit != nil {
		clusterType = cvt.To.ConfigCenter.ClusterType(newChange.TargetTenantInit.ClusterType)
	} else if newChange.TargetTemplate != nil {
		clusterType = cvt.To.ConfigCenter.ClusterType(newChange.TargetTemplate.ClusterType)
	}

	// evaluate apply policy
	evaluator := NewParameterApplyPolicyEvaluator(param.ChangeType)
	for _, fileChange := range newChange.Scope.Files {
		filename := fileChange.Filename
		configType := fileChange.ConfigType

		if len(fileChange.ModifiedParameters) == 0 {
			continue
		}

		// get attribute from config dict
		targetVersion := builder.targetVersionForFile(filename)
		listDictReq := &ccproto.ListConfigurationDictItemsRequest{
			ClusterType:         clusterType,
			ConfigurationType:   &configType,
			Version:             &targetVersion,
			ParamKeyIn:          container.TransformSlice(fileChange.ModifiedParameters, func(p ccent.ChangedParameter) string { return p.Name }),
			MatchPartialVersion: true,

			Page:     1,
			PageSize: int64(len(fileChange.ModifiedParameters) * 2), // double the size to avoid pagination
		}
		listDictResp, err := a.ConfigCenter.ConfigAPI.ListConfigurationDictItems(ctx, listDictReq)
		if err != nil {
			return nil, fmt.Errorf("failed to list configuration dict items: %v", err)
		}

		// trans to map
		attrs := map[string]ccent.ParameterAttribute{}
		for _, dictItem := range listDictResp.ConfigurationDictItems {
			attrs[dictItem.ParamKey] = *cvt.From.ConfigCenter.ParameterAttribute(dictItem)
		}

		// evaluate file
		evaluator.EvaluateFile(filename, configType, fileChange.ModifiedParameters, attrs)
	}

	result.ApplyPolicies, result.Reason = evaluator.Result()

	return result, nil
}

func (a *actorCreate) Create(param api.CreateConfigurationChangeParam) error {
	return a.create(param, nil)
}

func (a *actorCreate) CreateManaged(param api.CreateConfigurationChangeParam, ref api.ManagedConfigurationChangeRef) error {
	return a.create(param, &ref)
}

func (a *actorCreate) create(param api.CreateConfigurationChangeParam, managedRef *api.ManagedConfigurationChangeRef) error {
	ctx := a.Context

	builder, err := a.newConfigurationChange(param)
	if err != nil {
		return err
	}
	newChange := builder.Change

	err = a.creatable(newChange)
	if err != nil {
		return err
	}

	err = a.validateCreate(newChange)
	if err != nil {
		return err
	}
	if managedRef != nil {
		if managedRef.StepID == 0 {
			return errs.APIErrorf(baseapi.ErrCodeInvalidArgument, "managed configuration change step reference is required")
		}
		newChange.Phase = ccent.ConfigurationChangePhaseWaitForApply
		newChange.ApprovalID = nil
		newChange.PremiumUpgradeStepID = pointer.Pointer(managedRef.StepID)
	}

	if param.ExecutionWindow != nil {
		window, err := a.MaintenanceSvc.CreateOpsExecutionWindow(ctx, *param.ExecutionWindow)
		if err != nil {
			return err
		}
		newChange.ExecutionWindowID = pointer.Pointer(window.ID)
		newChange.ExecutionWindow = window
	}

	// save a new record to repo
	createdChange, err := a.Repo.CreateConfigurationChange(ctx, newChange)
	if err != nil {
		return err
	}

	a.Change = createdChange
	return nil
}

func (a *actorCreate) newConfigurationChange(param api.CreateConfigurationChangeParam) (*ConfigurationChangeBuilder, error) {
	builder, err := newConfigurationChangeBuilder(a.Context, a.Components, param)
	if err != nil {
		return nil, err
	}

	err = builder.AssembleConfiguration()
	if err != nil {
		return nil, err
	}

	return builder, nil
}

type ConfigurationChangeBuilder struct {
	// Inputs
	Context context.Context
	Comps   Components
	Param   api.CreateConfigurationChangeParam

	// Outputs
	Change        *ccent.ConfigurationChange
	CurrentConfig *ccent.Configuration
	ChangedConfig *ccent.Configuration

	// fileTargetVersions records the version whose parameter dictionary should
	// be used for an automatically selected configuration template. It differs
	// from Change.TargetVersion during a component-level upgrade.
	fileTargetVersions map[string]string
}

func newConfigurationChangeBuilder(ctx context.Context, comps Components, param api.CreateConfigurationChangeParam) (*ConfigurationChangeBuilder, error) {
	var err error
	b := &ConfigurationChangeBuilder{
		Context:            ctx,
		Comps:              comps,
		Param:              filterParamByChangeType(param),
		fileTargetVersions: make(map[string]string),
	}

	b.Change = &ccent.ConfigurationChange{
		ID:             0,
		Reason:         param.Reason,
		Applicant:      param.Applicant,
		OperatorEntity: param.OperatorEntity,
		ApplyPolicy:    b.Param.ApplyPolicy,
		Phase:          ccent.ConfigurationChangePhaseCreated,
	}

	if b.Param.TargetCluster != nil {
		b.CurrentConfig, err = b.Comps.ConfigService.GetClusterConfiguration(ctx, *b.Param.TargetCluster, param.ChangeType.MainComponent())
		if err != nil {
			return nil, fmt.Errorf("get current configuration failed: %w", err)
		}

		for c := range b.CurrentConfig.Components {
			if !param.ChangeType.ComponentExist(c) {
				delete(b.CurrentConfig.Components, c)
			}
		}
		if b.Param.ChangeType == api.ChangeTypeDm {
			if _, ok := b.CurrentConfig.Components[ccproto.DEDICATED_TIDB_COMPONENT_DM]; !ok {
				b.CurrentConfig.Components[ccproto.DEDICATED_TIDB_COMPONENT_DM] = &ccent.Component{
					Name:              ccproto.DEDICATED_TIDB_COMPONENT_DM,
					BaseImage:         "UNSPECIFIED",
					EffectiveImageTag: "UNSPECIFIED",
				}
			}
		}

		for filename, fVal := range b.CurrentConfig.Files {
			if !param.ChangeType.ConfigTypeExist(fVal.ConfigType) {
				delete(b.CurrentConfig.Files, filename)
			}
		}

		b.Change.TargetType = ccent.ConfigurationChangeTargetTypeCluster
		b.Change.TargetCluster = b.Param.TargetCluster
		b.Change.TargetVersion = b.CurrentConfig.Version
		b.Change.ChangeType = string(b.Param.ChangeType)
	}
	if b.Param.TargetTenantInit != nil {
		b.CurrentConfig, err = b.Comps.ConfigService.GetTenantInitConfiguration(ctx, *b.Param.TargetTenantInit)
		if err != nil {
			return nil, fmt.Errorf("get current configuration failed: %w", err)
		}

		b.Change.TargetType = ccent.ConfigurationChangeTargetTypeTenantInit
		b.Change.TargetTenantInit = b.Param.TargetTenantInit
		b.Change.TargetVersion = b.CurrentConfig.Version
		b.Change.ChangeType = string(b.Param.ChangeType)
	}
	if b.Param.TargetTemplate != nil {
		// only support create now, so current config is empty
		b.CurrentConfig = &ccent.Configuration{
			Version:    "",
			Components: nil,
			Files:      map[string]*ccent.FileConfiguration{},
		}
		b.Change.TargetTemplate = b.Param.TargetTemplate
		b.Change.TargetType = ccent.ConfigurationChangeTargetTypeTemplate
		b.Change.ChangeType = string(b.Param.ChangeType)
	}

	b.ChangedConfig, err = pointer.DeepCopy(b.CurrentConfig) // be same as current config and modify it when assemble
	if err != nil {
		return nil, fmt.Errorf("deep copy current configuration failed: %s", err)
	}

	return b, nil
}

func filterParamByChangeType(param api.CreateConfigurationChangeParam) api.CreateConfigurationChangeParam {
	filtedComps := make([]*api.ChangeComponentParam, 0)
	filtedConfigs := make([]*api.ChangeFileParam, 0)
	for _, comp := range param.Components {
		if param.ChangeType.ComponentExist(comp.Component) {
			filtedComps = append(filtedComps, comp)
		}
	}
	for _, file := range param.Files {
		if param.ChangeType.ConfigTypeExist(file.ConfigType) {
			filtedConfigs = append(filtedConfigs, file)
		}
	}
	param.Components = filtedComps
	param.Files = filtedConfigs
	return param
}

func (b *ConfigurationChangeBuilder) AssembleConfiguration() error {
	// assemble version change
	if b.Param.Version != nil {
		err := b.AssembleVersionChange(b.Param.Version)
		if err != nil {
			return fmt.Errorf("assemble change of version failed: %w", err)
		}
	}

	// assemble component change
	for _, param := range b.Param.Components {
		err := b.AssembleComponentChange(param, b.Param.Version)
		if err != nil {
			return fmt.Errorf("assemble change of component %s failed: %w", param.Component, err)
		}
	}

	// assemble file change
	for _, param := range b.Param.Files {
		if b.Param.TargetCluster != nil && b.CurrentConfig.Files[param.Filename] == nil {
			return fmt.Errorf("File %s does not exist and cannot be modified; please update the initial configuration via the Ops Tenant page instead.", param.Filename)
		}
		err := b.AssembleFileChange(param)
		if err != nil {
			return fmt.Errorf("assemble change of file %s failed: %w", param.Filename, err)
		}
	}

	return nil
}

func supportComponentLevelVersion(changeType api.ChangeType) bool {
	return changeType == api.ChangeTypeLogicalCluster || changeType == api.ChangeTypeSharedPool
}

func (b *ConfigurationChangeBuilder) AssembleComponentChange(param *api.ChangeComponentParam, version *api.ChangeVersionParam) error {
	comp := param.Component
	componentVersion := version
	hasComponentLevelVersion := false
	if param.Version != nil {
		if !supportComponentLevelVersion(b.Param.ChangeType) {
			return fmt.Errorf("component-level version is only supported for logicalcluster and sharedpool changes")
		}
		if version != nil && version.ToVersion != param.Version.ToVersion {
			return fmt.Errorf("component-level version %s conflicts with global version %s", param.Version.ToVersion, version.ToVersion)
		}
		componentVersion = param.Version
		hasComponentLevelVersion = true
	}

	curComponent := b.CurrentConfig.Components[comp]
	chgComponent := b.ChangedConfig.Components[comp]
	if curComponent == nil {
		return fmt.Errorf("component %s is not existing now so can't modify it", param.Component)
	}
	compChange := &ccent.ConfigurationComponentChange{
		ID:        0,
		Component: comp,
	}

	// build overwrite image tag change
	if param.OverwriteImageTag != nil && curComponent.OverwriteImageTag != param.OverwriteImageTag.ToImageTag {
		compChange.ModifiedOverwriteImageTag = &ccent.ChangedImageTag{
			From: curComponent.OverwriteImageTag,
			To:   param.OverwriteImageTag.ToImageTag,
		}
		chgComponent.OverwriteImageTag = param.OverwriteImageTag.ToImageTag
	}

	// build component version change
	if hasComponentLevelVersion && curComponent.Version != componentVersion.ToVersion {
		compChange.ModifiedVersion = &ccent.ChangedVersion{
			From: curComponent.Version,
			To:   componentVersion.ToVersion,
		}
		chgComponent.Version = componentVersion.ToVersion

		if b.Change.TargetCluster != nil {
			partialComponents := map[string]struct{}{comp: {}}
			if err := b.addVersionTemplateFileChanges(componentVersion.ToVersion, partialComponents); err != nil {
				return err
			}
		}
	}

	// build image tag change
	curImageTag := curComponent.EffectiveImageTag
	chgImageTag := chgComponent.OverwriteImageTag

	if componentVersion == nil {
		// patch cluster image
		if chgImageTag != "" && chgImageTag != curImageTag {
			compChange.ModifiedImageTag = &ccent.ChangedImageTag{
				From: curImageTag,
				To:   chgImageTag,
			}
			chgComponent.EffectiveImageTag = chgImageTag
		}
	} else {
		// upgrade cluster
		if param.OverwriteImageTag == nil {
			if curComponent.OverwriteImageTag != "" {
				compChange.ModifiedOverwriteImageTag = &ccent.ChangedImageTag{
					From: curComponent.OverwriteImageTag,
					To:   "",
				}
			}
			chgComponent.OverwriteImageTag = ""
			chgImageTag = ""
		}
		if chgImageTag == "" {
			imgReasp, err := b.Comps.ConfigCenter.ConfigAPI.GetComponentImage(b.Context, &ccproto.GetComponentImageRequest{
				ClusterType:      cvt.To.ConfigCenter.ClusterType(b.Change.TargetCluster.ClusterType),
				ComponentName:    comp,
				ComponentVersion: componentVersion.ToVersion,
			})
			if err != nil {
				return fmt.Errorf("get component image for %s failed: %w", comp, err)
			}
			chgImageTag = imgReasp.Tag
		}
		if chgImageTag != curImageTag {
			compChange.ModifiedImageTag = &ccent.ChangedImageTag{
				From: curImageTag,
				To:   chgImageTag,
			}
			chgComponent.EffectiveImageTag = chgImageTag
		}
	}

	if compChange.ChangedScope().Empty() {
		// no change for this file, so do not record it in change.
		return nil
	}

	b.Change.Scope.Components = append(b.Change.Scope.Components, compChange)
	b.ChangedConfig.Components[comp] = chgComponent

	return nil
}

// AssembleFileChange modify `b.Change` and b.ChangedConfig by file change.
func (b *ConfigurationChangeBuilder) AssembleFileChange(param *api.ChangeFileParam) error {
	ctx := b.Context
	filename := param.Filename
	configType := param.ConfigType

	curFileConfig := b.CurrentConfig.Files[filename]
	chgFileConfig := b.ChangedConfig.Files[filename]
	if curFileConfig == nil { // modified file is new, so make current config empty.
		curFileConfig = &ccent.FileConfiguration{}
		chgFileConfig = &ccent.FileConfiguration{}
	}
	fileChange := &ccent.ConfigurationFileChange{
		ID:           0,
		Filename:     filename,
		ConfigType:   configType,
		TemplateID:   curFileConfig.TemplateID,
		PreviousText: param.PreviousText,
	}

	if curFileConfig.Text != fileChange.PreviousText {
		return fmt.Errorf("%s previous text is unmatched with current text", filename)
	}

	// build modified template
	if param.Template != nil && curFileConfig.TemplateID != param.Template.ToTemplateID {
		fileChange.ModifiedTemplate = &ccent.ChangedTemplate{
			From: curFileConfig.TemplateID,
			To:   param.Template.ToTemplateID,
		}
		fileChange.TemplateID = param.Template.ToTemplateID
		chgFileConfig.TemplateID = param.Template.ToTemplateID
	}

	// build modified parameters
	builer := common.NewUnifiedParameterBuilder()
	// add custom parameters
	for idx := range param.CustomParameters {
		builer.AddCustom(param.CustomParameters[idx])
	}
	// add template paramters
	if fileChange.TemplateID != 0 {
		err := builer.InjectTemplates(ctx, b.Comps.ConfigService, *b.Change.ClusterType(), fileChange.TemplateID)
		if err != nil {
			return err
		}
	}

	// FIXEME: directly get cluster type
	clusterType := ccproto.ClusterType_DEDICATED_TIDB
	if b.Change.TargetCluster != nil {
		clusterType = cvt.To.ConfigCenter.ClusterType(b.Change.TargetCluster.ClusterType)
	} else if b.Change.TargetTenantInit != nil {
		clusterType = cvt.To.ConfigCenter.ClusterType(b.Change.TargetTenantInit.ClusterType)
	} else if b.Change.TargetTemplate != nil {
		clusterType = cvt.To.ConfigCenter.ClusterType(b.Change.TargetTemplate.ClusterType)
	}

	// add paramter attribute
	targetVersion := b.targetVersionForFile(filename)
	err := builer.InjectAttrs(ctx, clusterType, configType, targetVersion, b.Comps.ConfigCenter)
	if err != nil {
		return err
	}

	// diff parameters
	modifParams := builer.Build()
	fileChange.ModifiedParameters = common.DiffParameters(curFileConfig.Parameters, modifParams)
	for _, modifParam := range modifParams {
		fileChange.AllParameters = append(fileChange.AllParameters, modifParam.UsedParameter)
	}
	chgFileConfig.Parameters = modifParams

	if fileChange.ChangedScope().Empty() {
		// no change for this file, so do not record it in change.
		return nil
	}

	// if custom param is modified, check whether need to clear all custom param
	modifiedCustomParam := false
	for _, modif := range fileChange.ModifiedParameters {
		if modif.FromSource == ccent.ParameterCustom {
			modifiedCustomParam = true
			break
		}
	}
	if modifiedCustomParam {
		fileChange.ClearCustomParams = true
		for _, p := range fileChange.AllParameters {
			if p.Source == ccent.ParameterCustom {
				fileChange.ClearCustomParams = false
				break
			}
		}
	}

	b.Change.Scope.Files = append(b.Change.Scope.Files, fileChange)
	b.ChangedConfig.Files[filename] = chgFileConfig
	return nil
}

// AssembleVersionChange modify `b.Change` and b.ChangedConfig by version change.
//
// NOTE: It will cause change for all file and components, so call it firstly.
func (b *ConfigurationChangeBuilder) AssembleVersionChange(param *api.ChangeVersionParam) error {
	b.Change.Scope.Version = &ccent.ConfigurationVersionChange{
		ID:   0,
		From: b.CurrentConfig.Version,
		To:   param.ToVersion,
	}
	b.Change.TargetVersion = param.ToVersion
	b.ChangedConfig.Version = param.ToVersion

	partialComponents := partialVersionChangeComponents(b.Change.TargetCluster, b.Param.Components)

	// version change will cause image change for components, so update the `b.Param`
	// `AssembleComponentChange` will handle thoese changes.
	for _, compName := range versionChangeComponentNames(b.CurrentConfig.Components, partialComponents) {
		idx, compParam := b.Param.FindComponent(compName)

		// if not modify this component, we need to set it to update overwite image tag
		if compParam == nil {
			compParam = &api.ChangeComponentParam{
				Component: compName,
			}
			b.Param.Components = append(b.Param.Components, compParam)
			idx = len(b.Param.Components) - 1
		}

		// if specify image, do not update it.
		if compParam.OverwriteImageTag != nil {
			continue
		}

		// set ovewrite image tag to "", so it is same as version.
		compParam.OverwriteImageTag = &api.ChangeImageTagParam{
			ToImageTag: "",
		}

		b.Param.Components[idx] = compParam
	}

	// version change will cause template change for all file, so update the `b.Param`
	// `AssembleFileChange` will handle thoese changes.
	if err := b.addVersionTemplateFileChanges(b.Change.TargetVersion, partialComponents); err != nil {
		return err
	}

	return nil
}

func (b *ConfigurationChangeBuilder) addVersionTemplateFileChanges(targetVersion string, partialComponents map[string]struct{}) error {
	ctx := b.Context
	if b.Change.TargetCluster == nil {
		return nil
	}

	// Version change will cause template change for selected files, so update the b.Param.
	// AssembleFileChange will handle those changes later.
	for filename, curFile := range b.CurrentConfig.Files {
		if !shouldUpdateFileTemplateForVersion(b.Change.TargetCluster.ClusterType, curFile, partialComponents) {
			continue
		}
		b.setFileTargetVersion(filename, targetVersion)

		idx, fileParam := b.Param.FindFile(filename)

		// If not modifying this file yet, add a file param to update its template.
		if fileParam == nil {
			fileParam = &api.ChangeFileParam{
				Filename:         filename,
				ConfigType:       curFile.ConfigType,
				CustomParameters: curFile.ExtractCustomParameters(),
				Template:         nil, // update later
				PreviousText:     curFile.Text,
			}

			b.Param.Files = append(b.Param.Files, fileParam)
			idx = len(b.Param.Files) - 1
		}

		// If the caller specified a template, do not override it.
		if fileParam.Template != nil {
			continue
		}

		// Find component to get instance type and target version.
		comp := findComponentForConfigFile(b.Change.TargetCluster.ClusterType, b.CurrentConfig.Components, curFile)
		if comp == nil {
			return fmt.Errorf("file %s is used but its component %s is not found in components", filename, curFile.BelongToComponent)
		}

		defCase := ccent.TemplateDefaultCase{
			CloudInstanceType: comp.CloudInstanceType,
			Version:           targetVersion,
			ConfigType:        curFile.ConfigType,
			TenantID:          "",
		}
		// Template for tenant level.
		switch b.Change.TargetCluster.ClusterType {
		case ccent.ClusterTypeDedicatedTiDB:
			clusterIDInt, err := strconv.ParseUint(b.Change.TargetCluster.ClusterID, 10, 64)
			if err != nil {
				return fmt.Errorf("parse cluster id %s to uint64 failed: %s", b.Change.TargetCluster.ClusterID, err)
			}
			cluster, err := b.Comps.CoreSvc.Mgmt.GetCluster(ctx, clusterIDInt)
			if err != nil {
				return fmt.Errorf("get cluster %s failed: %s", b.Change.TargetCluster.ClusterID, err)
			}
			defCase.TenantID = strconv.FormatUint(cluster.OrganizationId, 10)
		}

		// Search new template for target version.
		newTemplate, err := b.Comps.ConfigService.GetTemplateByDefaultCase(ctx,
			b.CurrentConfig.TargetCluster.ClusterType,
			defCase,
		)
		if err != nil {
			return fmt.Errorf("not found new template when change version for file %s and version %s: %s",
				filename, targetVersion, err)
		}

		b.Param.Files[idx].Template = &api.ChangeTemplateParam{
			ToTemplateID: newTemplate.ID,
		}
	}

	return nil
}

func (b *ConfigurationChangeBuilder) setFileTargetVersion(filename, version string) {
	if b.fileTargetVersions == nil {
		b.fileTargetVersions = make(map[string]string)
	}
	b.fileTargetVersions[filename] = version
}

func (b *ConfigurationChangeBuilder) targetVersionForFile(filename string) string {
	if version := b.fileTargetVersions[filename]; version != "" {
		return version
	}
	return b.Change.TargetVersion
}

func partialVersionChangeComponents(target *ccent.ClusterConfigurationLocator, params []*api.ChangeComponentParam) map[string]struct{} {
	if target == nil || len(params) == 0 {
		return nil
	}
	switch target.ClusterType {
	case ccent.ClusterTypeLogicalCluster, ccent.ClusterTypeSharedPool:
	default:
		return nil
	}

	components := make(map[string]struct{}, len(params))
	for _, param := range params {
		if param.Component == "" {
			continue
		}
		components[param.Component] = struct{}{}
	}
	if len(components) == 0 {
		return nil
	}
	return components
}

func versionChangeComponentNames(current map[string]*ccent.Component, partialComponents map[string]struct{}) []string {
	components := make([]string, 0, len(current))
	if len(partialComponents) == 0 {
		for component := range current {
			components = append(components, component)
		}
		return components
	}

	for component := range partialComponents {
		if current[component] != nil {
			components = append(components, component)
		}
	}
	return components
}

func shouldUpdateFileTemplateForVersion(clusterType ccent.ClusterType, file *ccent.FileConfiguration, partialComponents map[string]struct{}) bool {
	if len(partialComponents) == 0 {
		return true
	}
	for _, component := range componentCandidatesForConfigFile(clusterType, file) {
		if _, ok := partialComponents[component]; ok {
			return true
		}
	}
	return false
}

func findComponentForConfigFile(clusterType ccent.ClusterType, components map[string]*ccent.Component, file *ccent.FileConfiguration) *ccent.Component {
	for _, component := range componentCandidatesForConfigFile(clusterType, file) {
		if comp := components[component]; comp != nil {
			return comp
		}
	}
	return nil
}

func componentCandidatesForConfigFile(clusterType ccent.ClusterType, file *ccent.FileConfiguration) []string {
	candidates := make([]string, 0, 3)
	appendCandidate := func(component string) {
		if component == "" || slices.Contains(candidates, component) {
			return
		}
		candidates = append(candidates, component)
	}

	appendCandidate(file.BelongToComponent)
	appendCandidate(cvt.To.ConfigCenter.ClusterType(clusterType).ConfigType2Component(file.ConfigType))
	appendCandidate(file.ConfigType) // compatibility for old files whose belong_to_component is empty

	return candidates
}

func (a *actorCreate) creatable(newChange *ccent.ConfigurationChange) error {
	ctx := a.Context

	// changeable
	if newChange.TargetCluster != nil {
		reason, err := ClusterConfigurationChangeableForChange(ctx, a.Components, newChange)
		if err != nil {
			return err
		}
		if reason != "" {
			return errs.APIErrorNew(baseapi.ErrCodeFailedPrecond, errors.New(reason))
		}
	}
	if newChange.TargetTenantInit != nil {
		reason, err := TenantInitConfigurationChangeable(ctx, a.Components, *newChange.TargetTenantInit)
		if err != nil {
			return err
		}
		if reason != "" {
			return errs.APIErrorNew(baseapi.ErrCodeFailedPrecond, errors.New(reason))
		}
	}

	return nil
}

func (a *actorCreate) validateCreate(newChange *ccent.ConfigurationChange) error {
	ctx := a.Context

	clusterType := ccent.ClusterTypeDedicatedTiDB
	if newChange.TargetCluster != nil {
		clusterType = newChange.TargetCluster.ClusterType
	} else if newChange.TargetTenantInit != nil {
		clusterType = newChange.TargetTenantInit.ClusterType
	} else if newChange.TargetTemplate != nil {
		clusterType = newChange.TargetTemplate.ClusterType
	}

	for _, component := range newChange.Scope.Components {
		targetVersion := ""
		if newChange.Scope.Version != nil {
			targetVersion = newChange.Scope.Version.To
		}
		if component.ModifiedVersion != nil {
			targetVersion = component.ModifiedVersion.To
		}
		if targetVersion == "" {
			continue
		}

		versions, err := a.ConfigService.ListAvailableComponentVersions(ctx, clusterType, component.Component)
		if err != nil {
			return fmt.Errorf("list available version failed: %s", err)
		}
		if !slices.Contains(versions, targetVersion) {
			return fmt.Errorf("modified version %s is not found", targetVersion)
		}
	}

	return nil
}