package tioperator

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"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"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	tikvWorkerComponentLabel        = "tikv-worker"
	coprocessorWorkerComponentLabel = "coprocessor-worker"
	schemaManagerComponentLabel     = "schema-manager"
	tikvWorkerSuffix                = "-default"
	coprocessorWorkerSuffix         = "-coprocessor"
	schemaManagerSuffix             = "-schema-manager"
)

func ListTiKVWorkers(ctx context.Context, remoteCli client.Client, clusterId string) ([]*entity.Component, error) {
	return listTiKVWorkerGroups(ctx, remoteCli, dataplaneNamespace(clusterId))
}

func GetTiKVWorker(ctx context.Context, remoteCli client.Client, clusterId, groupId string) (*entity.Component, error) {
	tikvWorkerCr, err := getTiKVWorkerGroupCR(ctx, remoteCli, dataplaneNamespace(clusterId), groupId)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "tikv worker %s not found: %s", groupId, err)
		}
		return nil, status.Errorf(codes.Internal, "failed to get tikv worker %s: %s", groupId, err)
	}

	return cvtTiKVWorkerGroupToEn(ctx, tikvWorkerCr)
}

func UpdateTiKVWorker(ctx context.Context, remoteCli client.Client, clusterId, groupId string, options []UpdateOption) (*entity.Component, error) {
	tikvWorkerCr, err := getTiKVWorkerGroupCR(ctx, remoteCli, dataplaneNamespace(clusterId), groupId)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "tikv worker %s not found: %s", groupId, err)
		}
		return nil, status.Errorf(codes.Internal, "failed to get tikv worker %s: %s", groupId, err)
	}

	needUpdate, err := getUpdateExpectedTiKVWorkerGroup(ctx, tikvWorkerCr, apiv1.ComponentType_TIKV_WORKER, options...)
	if err != nil {
		return nil, fmt.Errorf("failed to get expected tikv worker %s: %w", groupId, err)
	}

	if needUpdate {
		if err := remoteCli.Update(ctx, tikvWorkerCr); err != nil {
			return nil, status.Errorf(codes.Internal, "failed to update tikv worker %s: %s", groupId, err)
		}
	}

	return cvtTiKVWorkerGroupToEn(ctx, tikvWorkerCr)
}

func ListCoprocessorWorkers(ctx context.Context, remoteCli client.Client, clusterId string) ([]*entity.Component, error) {
	return listCoprocessorWorkerGroups(ctx, remoteCli, dataplaneNamespace(clusterId))
}

func GetCoprocessorWorker(ctx context.Context, remoteCli client.Client, clusterId, groupId string) (*entity.Component, error) {
	coprWorkerCr, err := getCoprocessorWorkerGroupCR(ctx, remoteCli, dataplaneNamespace(clusterId), groupId)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "coprocessor worker %s not found: %s", groupId, err)
		}
		return nil, status.Errorf(codes.Internal, "failed to get coprocessor worker %s: %s", groupId, err)
	}

	return cvtTiKVWorkerGroupToEn(ctx, coprWorkerCr)
}

func UpdateCoprocessorWorker(ctx context.Context, remoteCli client.Client, clusterId, groupId string, options []UpdateOption) (*entity.Component, error) {
	coprWorkerCr, err := getCoprocessorWorkerGroupCR(ctx, remoteCli, dataplaneNamespace(clusterId), groupId)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "coprocessor worker %s not found: %s", groupId, err)
		}
		return nil, status.Errorf(codes.Internal, "failed to get coprocessor worker %s: %s", groupId, err)
	}

	needUpdate, err := getUpdateExpectedTiKVWorkerGroup(ctx, coprWorkerCr, apiv1.ComponentType_COPROCESSOR_WORKER, options...)
	if err != nil {
		return nil, fmt.Errorf("failed to get expected coprocessor worker %s: %w", groupId, err)
	}

	if needUpdate {
		if err := remoteCli.Update(ctx, coprWorkerCr); err != nil {
			return nil, status.Errorf(codes.Internal, "failed to update coprocessor worker %s: %s", groupId, err)
		}
	}

	return cvtTiKVWorkerGroupToEn(ctx, coprWorkerCr)
}

func ListSchemaManagers(ctx context.Context, remoteCli client.Client, clusterId string) ([]*entity.Component, error) {
	return listSchemaManagerGroups(ctx, remoteCli, dataplaneNamespace(clusterId))
}

func GetSchemaManager(ctx context.Context, remoteCli client.Client, clusterId, groupId string) (*entity.Component, error) {
	schemaManagerCr, err := getSchemaManagerGroupCR(ctx, remoteCli, dataplaneNamespace(clusterId), groupId)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "schema manager %s not found: %s", groupId, err)
		}
		return nil, status.Errorf(codes.Internal, "failed to get schema manager %s: %s", groupId, err)
	}

	return cvtTiKVWorkerGroupToEn(ctx, schemaManagerCr)
}

func UpdateSchemaManager(ctx context.Context, remoteCli client.Client, clusterId, groupId string, options []UpdateOption) (*entity.Component, error) {
	schemaManagerCr, err := getSchemaManagerGroupCR(ctx, remoteCli, dataplaneNamespace(clusterId), groupId)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "schema manager %s not found: %s", groupId, err)
		}
		return nil, status.Errorf(codes.Internal, "failed to get schema manager %s: %s", groupId, err)
	}

	needUpdate, err := getUpdateExpectedTiKVWorkerGroup(ctx, schemaManagerCr, apiv1.ComponentType_SCHEMA_MANAGER, options...)
	if err != nil {
		return nil, fmt.Errorf("failed to get expected schema manager %s: %w", groupId, err)
	}

	if needUpdate {
		if err := remoteCli.Update(ctx, schemaManagerCr); err != nil {
			return nil, status.Errorf(codes.Internal, "failed to update schema manager %s: %s", groupId, err)
		}
	}

	return cvtTiKVWorkerGroupToEn(ctx, schemaManagerCr)
}

// getTiKVWorkerGroupCR gets TiKVWorkerGroup CR for tikv worker
// tikv worker group name: db-xxxx-default
func getTiKVWorkerGroupCR(ctx context.Context, remoteCli client.Client, dataplaneNs, groupId string) (*v1alpha1.TiKVWorkerGroup, error) {
	if !strings.HasPrefix(groupId, componentGroupPrefix) {
		groupId = fmt.Sprintf("%s%s", componentGroupPrefix, groupId)
	}

	// coprocessor worker group name has suffix "-default"
	groupName := groupId + tikvWorkerSuffix

	var tikvWorkerGroup v1alpha1.TiKVWorkerGroup
	key := types.NamespacedName{Name: groupName, Namespace: dataplaneNs}
	if err := remoteCli.Get(ctx, key, &tikvWorkerGroup); err != nil {
		return nil, err
	}

	// Verify it's a tikv-worker component
	if tikvWorkerGroup.Labels[schemav1.LabelComponentKey] != tikvWorkerComponentLabel {
		return nil, apierrors.NewNotFound(v1alpha1.Resource("tikvworkergroups"), groupId)
	}

	return &tikvWorkerGroup, nil
}

// getCoprocessorWorkerGroupCR gets TiKVWorkerGroup CR for coprocessor worker
// coprocessor worker group name: db-xxxx-coprocessor
func getCoprocessorWorkerGroupCR(ctx context.Context, remoteCli client.Client, dataplaneNs, groupId string) (*v1alpha1.TiKVWorkerGroup, error) {
	if !strings.HasPrefix(groupId, componentGroupPrefix) {
		groupId = fmt.Sprintf("%s%s", componentGroupPrefix, groupId)
	}
	// coprocessor worker group name has suffix "-coprocessor"
	groupName := groupId + coprocessorWorkerSuffix

	var coprWorkerGroup v1alpha1.TiKVWorkerGroup
	key := types.NamespacedName{Name: groupName, Namespace: dataplaneNs}
	if err := remoteCli.Get(ctx, key, &coprWorkerGroup); err != nil {
		return nil, err
	}

	// Verify it's a coprocessor-worker component
	if coprWorkerGroup.Labels[schemav1.LabelComponentKey] != coprocessorWorkerComponentLabel {
		return nil, apierrors.NewNotFound(v1alpha1.Resource("tikvworkergroups"), groupName)
	}

	return &coprWorkerGroup, nil
}

// getSchemaManagerGroupCR gets TiKVWorkerGroup CR for schema manager.
// schema manager group name: db-xxxx-schema-manager
func getSchemaManagerGroupCR(ctx context.Context, remoteCli client.Client, dataplaneNs, groupId string) (*v1alpha1.TiKVWorkerGroup, error) {
	if !strings.HasPrefix(groupId, componentGroupPrefix) {
		groupId = fmt.Sprintf("%s%s", componentGroupPrefix, groupId)
	}

	groupName := groupId + schemaManagerSuffix

	var schemaManagerGroup v1alpha1.TiKVWorkerGroup
	key := types.NamespacedName{Name: groupName, Namespace: dataplaneNs}
	if err := remoteCli.Get(ctx, key, &schemaManagerGroup); err != nil {
		return nil, err
	}

	if schemaManagerGroup.Labels[schemav1.LabelComponentKey] != schemaManagerComponentLabel {
		return nil, apierrors.NewNotFound(v1alpha1.Resource("tikvworkergroups"), groupId)
	}

	return &schemaManagerGroup, nil
}

func listTiKVWorkerGroups(ctx context.Context, remoteClient client.Client, dataplaneNs string) ([]*entity.Component, error) {
	var tikvWorkerGroups v1alpha1.TiKVWorkerGroupList
	if err := remoteClient.List(
		ctx,
		&tikvWorkerGroups,
		client.InNamespace(dataplaneNs),
		client.MatchingLabels{
			schemav1.LabelComponentKey: tikvWorkerComponentLabel,
		},
	); err != nil {
		return nil, status.Errorf(codes.Internal, "failed to list tikv-worker groups: %s", err)
	}

	tikvWorkerComponents := make([]*entity.Component, 0, len(tikvWorkerGroups.Items))
	for i := range tikvWorkerGroups.Items {
		compEn, err := cvtTiKVWorkerGroupToEn(ctx, &tikvWorkerGroups.Items[i])
		if err != nil {
			tilog.Warn(ctx, "convert tikv worker group to component", zap.Error(err))
		} else {
			tikvWorkerComponents = append(tikvWorkerComponents, compEn)
		}
	}
	return tikvWorkerComponents, nil
}

func listCoprocessorWorkerGroups(ctx context.Context, remoteClient client.Client, dataplaneNs string) ([]*entity.Component, error) {
	var tikvWorkerGroups v1alpha1.TiKVWorkerGroupList
	if err := remoteClient.List(
		ctx,
		&tikvWorkerGroups,
		client.InNamespace(dataplaneNs),
		client.MatchingLabels{
			schemav1.LabelComponentKey: coprocessorWorkerComponentLabel,
		},
	); err != nil {
		return nil, status.Errorf(codes.Internal, "failed to list coprocessor-worker groups: %s", err)
	}

	coprWorkerComponents := make([]*entity.Component, 0, len(tikvWorkerGroups.Items))
	for i := range tikvWorkerGroups.Items {
		compEn, err := cvtTiKVWorkerGroupToEn(ctx, &tikvWorkerGroups.Items[i])
		if err != nil {
			tilog.Warn(ctx, "convert coprocessor worker group to component", zap.Error(err))
		} else {
			coprWorkerComponents = append(coprWorkerComponents, compEn)
		}
	}
	return coprWorkerComponents, nil
}

func listSchemaManagerGroups(ctx context.Context, remoteClient client.Client, dataplaneNs string) ([]*entity.Component, error) {
	var tikvWorkerGroups v1alpha1.TiKVWorkerGroupList
	if err := remoteClient.List(
		ctx,
		&tikvWorkerGroups,
		client.InNamespace(dataplaneNs),
		client.MatchingLabels{
			schemav1.LabelComponentKey: schemaManagerComponentLabel,
		},
	); err != nil {
		return nil, status.Errorf(codes.Internal, "failed to list schema-manager groups: %s", err)
	}

	schemaManagerComponents := make([]*entity.Component, 0, len(tikvWorkerGroups.Items))
	for i := range tikvWorkerGroups.Items {
		compEn, err := cvtTiKVWorkerGroupToEn(ctx, &tikvWorkerGroups.Items[i])
		if err != nil {
			tilog.Warn(ctx, "convert schema manager group to component", zap.Error(err))
		} else {
			schemaManagerComponents = append(schemaManagerComponents, compEn)
		}
	}
	return schemaManagerComponents, nil
}

func cvtTiKVWorkerGroupToEn(ctx context.Context, tikvWorkerGroup *v1alpha1.TiKVWorkerGroup) (*entity.Component, error) {
	// Extract component ID from group name
	// tikv worker: db-<Id>-default -> Id
	// coprocessor worker: db-<Id>-coprocessor -> Id
	nameParts := strings.Split(tikvWorkerGroup.Name, "-")
	compId := nameParts[1]

	compEn := &entity.Component{
		Id: compId,
	}
	setTemplateVersionFromAnnotations(compEn, tikvWorkerGroup.Annotations)

	// Determine component type from labels
	switch tikvWorkerGroup.Labels[schemav1.LabelComponentKey] {
	case tikvWorkerComponentLabel:
		compEn.ComponentType = apiv1.ComponentType_TIKV_WORKER
	case coprocessorWorkerComponentLabel:
		compEn.ComponentType = apiv1.ComponentType_COPROCESSOR_WORKER
	case schemaManagerComponentLabel:
		compEn.ComponentType = apiv1.ComponentType_SCHEMA_MANAGER
	default:
		compEn.ComponentType = apiv1.ComponentType_TIKV_WORKER
	}

	// Find tikv-worker container in overlay
	var workerContainer corev1.Container
	foundWorkerContainer := false
	if tikvWorkerGroup.Spec.Template.Spec.Overlay != nil &&
		tikvWorkerGroup.Spec.Template.Spec.Overlay.Pod != nil {
		for i := range tikvWorkerGroup.Spec.Template.Spec.Overlay.Pod.Spec.Containers {
			if tikvWorkerGroup.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i].Name == "tikv-worker" {
				workerContainer = tikvWorkerGroup.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i]
				foundWorkerContainer = true
				break
			}
		}
	}
	if !foundWorkerContainer {
		return nil, fmt.Errorf("container named 'tikv-worker' not found in TiKVWorkerGroup %s", tikvWorkerGroup.Name)
	}

	// Get nodeSelector from Overlay if present
	var nodeSelector map[string]string
	if tikvWorkerGroup.Spec.Template.Spec.Overlay != nil &&
		tikvWorkerGroup.Spec.Template.Spec.Overlay.Pod != nil {
		nodeSelector = tikvWorkerGroup.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector
	}

	var err error
	compEn.NodeQuota, err = CvtNodeSpecToQuota(
		tikvWorkerGroup.Labels,
		tikvWorkerGroup.Annotations,
		uint32(*tikvWorkerGroup.Spec.Replicas),
		workerContainer.Resources.Limits,
		nodeSelector,
	)
	if err != nil {
		return nil, fmt.Errorf("failed to get tikv worker group %s quota: %w", compEn.GetId(), err)
	}

	compEn.Version = tikvWorkerGroup.Spec.Template.Spec.Version

	imageRepo, imageTag := splitImageRepoTag(tikvWorkerGroup.Spec.Template.Spec.Image)
	compEn.Configuration = &entity.ComponentConfiguration{
		ImageRepo:                imageRepo,
		ImageTag:                 imageTag,
		ConfigurationFileContent: string(tikvWorkerGroup.Spec.Template.Spec.Config),
	}

	compEn.Phase = getTiKVWorkerGroupStatus(ctx, tikvWorkerGroup)

	if compEn.ComponentType == apiv1.ComponentType_COPROCESSOR_WORKER || compEn.ComponentType == apiv1.ComponentType_SCHEMA_MANAGER {
		compEn.Storages, err = cvtStoragesToEn(ctx, tikvWorkerGroup.Spec.Template.Spec.Volumes)
		if err != nil {
			return nil, fmt.Errorf("failed to convert coprocessor worker group %s storages: %w", compEn.GetId(), err)
		}
	}

	return compEn, nil
}

func getTiKVWorkerGroupStatus(ctx context.Context, tikvWorkerGroup *v1alpha1.TiKVWorkerGroup) apiv1.Component_Phase {
	if !isSynced(tikvWorkerGroup.Status.Conditions) {
		tilog.Warn(ctx, "tikv worker group is not synced", zap.String("name", tikvWorkerGroup.Name), zap.Any("Conditions", tikvWorkerGroup.Status.Conditions))
		return apiv1.Component_MODIFYING
	}

	if !isUpToDate(tikvWorkerGroup.Generation, tikvWorkerGroup.Status.CommonStatus) {
		tilog.Warn(ctx, "tikv worker group is not up to date", zap.String("name", tikvWorkerGroup.Name), zap.Any("Status", tikvWorkerGroup.Status))
		return apiv1.Component_MODIFYING
	}

	isSchemaManager := tikvWorkerGroup.Labels[schemav1.LabelComponentKey] == schemaManagerComponentLabel
	if (tikvWorkerGroup.Status.ReadyReplicas == 0 || tikvWorkerGroup.Status.UpdatedReplicas == 0) &&
		(!isSchemaManager || lo.FromPtr(tikvWorkerGroup.Spec.Replicas) > 0) {
		tilog.Warn(ctx, "tikv worker group has zero ready or updated replicas",
			zap.String("name", tikvWorkerGroup.Name),
			zap.Int32("ReadyReplicas", tikvWorkerGroup.Status.ReadyReplicas),
			zap.Int32("UpdatedReplicas", tikvWorkerGroup.Status.UpdatedReplicas))
		return apiv1.Component_UNAVAILABLE
	}

	// check if tikv worker group is ready
	if !isReady(tikvWorkerGroup.Status.Conditions) {
		tilog.Warn(ctx, "tikv worker group is not ready", zap.String("name", tikvWorkerGroup.Name), zap.Any("Conditions", tikvWorkerGroup.Status.Conditions))
		return apiv1.Component_UNAVAILABLE
	}

	return apiv1.Component_AVAILABLE
}

func getUpdateExpectedTiKVWorkerGroup(ctx context.Context, tikvWorkerCr *v1alpha1.TiKVWorkerGroup, compType apiv1.ComponentType, options ...UpdateOption) (bool, error) {
	var needUpdate bool

	for _, opt := range options {
		switch opt.Option {
		case OpsUpdateNodeCount:
			count, ok := opt.Value.(uint32)
			if !ok {
				tilog.Error(ctx, "invalid node count", zap.Any("value", opt.Value))
				return false, fmt.Errorf("invalid node count for tikv worker %s", tikvWorkerCr.Name)
			}
			if count > 0 {
				currNodeQuota, err := CvtNodeSpecToQuota(tikvWorkerCr.Labels, tikvWorkerCr.Annotations, uint32(*tikvWorkerCr.Spec.Replicas), nil, nil)
				if err != nil {
					return false, status.Errorf(codes.Internal, "failed to convert node spec to quota for tikv worker %s: %s", tikvWorkerCr.Name, err)
				}

				nodeQuota := &entity.NodeQuota{
					NodeCount: count,
				}

				err = checkNodeQuota(currNodeQuota, nodeQuota)
				if err != nil {
					return false, status.Errorf(codes.InvalidArgument, "invalid node quota for tikv worker %s: %s", tikvWorkerCr.Name, err)
				}

				if uint32(*tikvWorkerCr.Spec.Replicas) != count {
					tikvWorkerCr.Spec.Replicas = lo.ToPtr(int32(count))
					needUpdate = true
				}
			} else if tikvWorkerCr.Labels[schemav1.LabelComponentKey] == schemaManagerComponentLabel {
				// Only schema manager is allowed to scale down to zero; other
				// tikv worker variants treat count==0 as a no-op.
				if uint32(*tikvWorkerCr.Spec.Replicas) != count {
					tikvWorkerCr.Spec.Replicas = lo.ToPtr(int32(count))
					needUpdate = true
				}
			}

		case OpsUpdateNodeSpec:
			nodeSpec, ok := opt.Value.(*entity.NodeSpec)
			if !ok {
				tilog.Error(ctx, "invalid node spec", zap.Any("value", opt.Value))
				return false, fmt.Errorf("invalid node spec for tikv worker %s", tikvWorkerCr.Name)
			}

			resourceRequirements, err := setNodeSpec(nodeSpec, tikvWorkerCr.Labels, tikvWorkerCr.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector, compType)
			if err != nil {
				tilog.Error(ctx, "set node spec", zap.Error(err))
				return false, fmt.Errorf("failed to set node spec for tikv worker %s: %w", tikvWorkerCr.Name, err)
			}
			for i := range tikvWorkerCr.Spec.Template.Spec.Overlay.Pod.Spec.Containers {
				if tikvWorkerCr.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i].Name == "tikv-worker" {
					tikvWorkerCr.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i].Resources = resourceRequirements
					needUpdate = true
					break
				}
			}
		case OpsUpdateMaxCount:
			maxCount, ok := opt.Value.(uint32)
			if !ok {
				tilog.Error(ctx, "invalid max count", zap.Any("value", opt.Value))
				return false, fmt.Errorf("invalid max count for tikv worker %s", tikvWorkerCr.Name)
			}
			if maxCount > 0 {
				annotations := tikvWorkerCr.Annotations
				if annotations == nil {
					annotations = map[string]string{}
				}

				annotations[schemav1.AnnotationsComponentMaxCount] = fmt.Sprintf("%d", maxCount)
				tikvWorkerCr.Annotations = annotations

				needUpdate = true
			}
		case OpsUpdateMinCount:
			minCount, ok := opt.Value.(uint32)
			if !ok {
				tilog.Error(ctx, "invalid min count", zap.Any("value", opt.Value))
			}

			annotations := tikvWorkerCr.Annotations
			if annotations == nil {
				annotations = map[string]string{}
			}

			annotations[schemav1.AnnotationsComponentMinCount] = fmt.Sprintf("%d", minCount)
			tikvWorkerCr.Annotations = annotations

			needUpdate = true
		case OpsUpdateVersion:
			version, ok := opt.Value.(string)
			if !ok {
				tilog.Error(ctx, "invalid version", zap.Any("value", opt.Value))
				return false, fmt.Errorf("invalid version for tikv worker %s", tikvWorkerCr.Name)
			}
			if version != "" {
				tikvWorkerCr.Spec.Template.Spec.Version = version
				needUpdate = true
			}
		case OpsUpdateImage:
			image, ok := opt.Value.(string)
			if !ok {
				tilog.Error(ctx, "invalid image", zap.Any("value", opt.Value))
				return false, fmt.Errorf("invalid image for tikv worker %s", tikvWorkerCr.Name)
			}
			if image != "" {
				tikvWorkerCr.Spec.Template.Spec.Image = &image
				needUpdate = true
			}
		case OpsUpdateConfig:
			config, ok := opt.Value.(string)
			if !ok {
				tilog.Error(ctx, "invalid config", zap.Any("value", opt.Value))
				return false, fmt.Errorf("invalid config for tikv worker %s", tikvWorkerCr.Name)
			}
			if config != "" {
				tikvWorkerCr.Spec.Template.Spec.Config = v1alpha1.ConfigFile(config)
				needUpdate = true
			}
		case OpsUpdateUpdateStrategy:
			strategy, ok := opt.Value.(v1alpha1.ConfigUpdateStrategy)
			if !ok {
				tilog.Error(ctx, "invalid update strategy", zap.Any("value", opt.Value))
				return false, fmt.Errorf("invalid update strategy for tikv worker %s", tikvWorkerCr.Name)
			}
			if strategy != "" {
				tikvWorkerCr.Spec.Template.Spec.UpdateStrategy.Config = strategy
				needUpdate = true
			}
		case OpsUpdateNodeVersion:
			nodeVersion, ok := opt.Value.(string)
			if !ok || nodeVersion == "" {
				tilog.Error(ctx, "invalid node version", zap.Any("value", opt.Value))
				return false, status.Errorf(codes.InvalidArgument, "invalid node version for tikv worker %s", tikvWorkerCr.Name)
			}
			nodeSelector := tikvWorkerCr.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector
			if nodeSelector == nil {
				nodeSelector = map[string]string{}
			}
			if nodeSelector[schemav1.K8sVersionLabelKey] != nodeVersion {
				nodeSelector[schemav1.K8sVersionLabelKey] = nodeVersion
				tikvWorkerCr.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector = nodeSelector
				needUpdate = true
			}
		}
	}

	// allow init volume for cop worker when not found
	// TODO: remove it later (after 2026-07-01) when all coprocessor worker has data volume initialized
	hasDataVolume := false
	for _, vol := range tikvWorkerCr.Spec.Template.Spec.Volumes {
		if vol.Name == string(schemav1.StorageData) {
			hasDataVolume = true
			break
		}
	}
	if !hasDataVolume {
		for _, opt := range options {
			if opt.Option != OpsUpdateDataVolumeAttributesClass {
				continue
			}

			volumeAttributesClass, ok := opt.Value.(string)
			if !ok {
				tilog.Error(ctx, "invalid data volume attributes class", zap.Any("value", opt.Value))
				return false, fmt.Errorf("invalid data volume attributes class for tikv worker %s", tikvWorkerCr.Name)
			}

			storageClassName := entity.CloudSSD
			tikvWorkerCr.Spec.Template.Spec.Volumes = append(tikvWorkerCr.Spec.Template.Spec.Volumes, v1alpha1.Volume{
				Name:                      string(schemav1.StorageData),
				Mounts:                    []v1alpha1.VolumeMount{{Type: v1alpha1.VolumeMountTypeTiKVData}},
				Storage:                   resource.MustParse("200Gi"),
				StorageClassName:          &storageClassName,
				VolumeAttributesClassName: &volumeAttributesClass,
			})

			componentLabel := tikvWorkerCr.Spec.Template.Spec.Overlay.Pod.Labels["tags.tidbcloud.com/component"]
			sharedPoolId := tikvWorkerCr.Labels[schemav1.LabelSharedPoolKey]
			userTenant := tikvWorkerCr.Spec.Template.Spec.Overlay.Pod.Labels[schemav1.LabelUserTenantKey]
			volumeClaimLabels := map[string]string{
				schemav1.LabelSharedPoolKey:    sharedPoolId,
				"tags.tidbcloud.com/component": componentLabel,
			}
			if userTenant != "" {
				volumeClaimLabels[schemav1.LabelUserTenantKey] = userTenant
			}
			if tikvWorkerCr.Spec.Template.Spec.Overlay == nil {
				tikvWorkerCr.Spec.Template.Spec.Overlay = &v1alpha1.Overlay{}
			}
			tikvWorkerCr.Spec.Template.Spec.Overlay.PersistentVolumeClaims = append(
				tikvWorkerCr.Spec.Template.Spec.Overlay.PersistentVolumeClaims,
				v1alpha1.NamedPersistentVolumeClaimOverlay{
					Name: string(schemav1.StorageData),
					PersistentVolumeClaim: v1alpha1.PersistentVolumeClaimOverlay{
						ObjectMeta: v1alpha1.ObjectMeta{
							Labels: volumeClaimLabels,
						},
					},
				},
			)
			needUpdate = true
			tilog.Info(ctx, "try to create volumn for cop worker", zap.Any("cop-worker", tikvWorkerCr.Spec))
			break
		}
	} else {
		// Handle data volume updates (for coprocessor worker)
		for i := range tikvWorkerCr.Spec.Template.Spec.Volumes {
			if tikvWorkerCr.Spec.Template.Spec.Volumes[i].Name == string(schemav1.StorageData) {
				modifyVolume := false
				for _, opt := range options {
					switch opt.Option {
					case OpsUpdateDataSize:
						size, ok := opt.Value.(uint32)
						if !ok {
							tilog.Error(ctx, "invalid data storage size", zap.Any("value", opt.Value))
							return false, fmt.Errorf("invalid data storage size for tikv worker %s", tikvWorkerCr.Name)
						}

						sizeQuantity := resource.MustParse(fmt.Sprintf("%dGi", size))
						if size > 0 && tikvWorkerCr.Spec.Template.Spec.Volumes[i].Storage.Cmp(sizeQuantity) < 0 {
							tikvWorkerCr.Spec.Template.Spec.Volumes[i].Storage = sizeQuantity
							modifyVolume = true
							needUpdate = true
						}

					case OpsUpdateDataVolumeAttributesClass:
						volumeAttributesClass, ok := opt.Value.(string)
						if !ok {
							tilog.Error(ctx, "invalid data volume attributes class", zap.Any("value", opt.Value))
							return false, fmt.Errorf("invalid data volume attributes class for tikv worker %s", tikvWorkerCr.Name)
						}
						curr := lo.FromPtr(tikvWorkerCr.Spec.Template.Spec.Volumes[i].VolumeAttributesClassName)
						if volumeAttributesClass != "" && volumeAttributesClass != curr {
							tikvWorkerCr.Spec.Template.Spec.Volumes[i].VolumeAttributesClassName = lo.ToPtr(strings.ToLower(volumeAttributesClass))
							modifyVolume = true
							needUpdate = true
						}
					}
				}
				if modifyVolume {
					if tikvWorkerCr.Annotations == nil {
						tikvWorkerCr.Annotations = make(map[string]string)
					}
					tikvWorkerCr.Annotations[schemav1.AnnotationsUpdateTimeDataVolumeKey] = time.Now().Format(time.RFC3339)
				}
			}
		}
	}

	return needUpdate, nil
}