package server

import (
	"context"
	"maps"
	"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"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	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"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
	"k8s.io/apimachinery/pkg/labels"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

func (s *serverImpl) getRunningDagIns(ctx context.Context, planNames, sharedPoolIds, tenantIds, tidbClusterIds, componentIds []string) (executor.ExecTaskIdStatus, error) {
	// Build a MatchFilter for server-side filtering instead of pulling all tasks
	filter := executor.NewMatchFilter().
		Add(workflow.SharedPoolIdKey, sharedPoolIds).
		Add(workflow.TenantIdKey, tenantIds).
		Add(workflow.TiDBClusterIdKey, tidbClusterIds).
		Add(workflow.ComponentIdsKey, componentIds)
	return s.queryRunningDagIns(ctx, planNames, filter)
}

func (s *serverImpl) queryRunningDagIns(ctx context.Context, planNames []string, filter executor.MatchFilter) (executor.ExecTaskIdStatus, error) {
	res := executor.NewExecTaskIdStatus()
	if len(planNames) == 0 {
		return res, nil
	}

	for _, planName := range planNames {
		dagIns, err := s.dagExecutor.QueryRunnings(ctx, planName, filter)
		if err != nil {
			return nil, err
		}

		maps.Copy(res, dagIns)
	}

	return res, nil
}

func (s *serverImpl) checkComponent(component *entity.Component) error {
	// tiflash-learner and tiflash-compute-learner are not supported
	if component.GetComponentType() == apiv1.ComponentType_TIFLASH_LEARNER ||
		component.GetComponentType() == apiv1.ComponentType_TIFLASH_COMPUTE_LEARNER {
		return nil
	}

	if component.GetComponentType() == apiv1.ComponentType_COMPONENT_TYPE_UNSPECIFIED {
		return status.Errorf(codes.InvalidArgument, "component %s-%s type is required", entity.GetComponentTypeName(component), component.GetId())
	}

	if component.GetId() == "" {
		return status.Errorf(codes.InvalidArgument, "component %s-%s id is required", entity.GetComponentTypeName(component), component.GetId())
	}

	if component.GetNodeQuota() == nil {
		return status.Errorf(codes.InvalidArgument, "component %s-%s node quota is required", entity.GetComponentTypeName(component), component.GetId())
	}

	if component.GetNodeQuota().GetNodeSpec() == nil {
		return status.Errorf(codes.InvalidArgument, "component %s-%s node spec is required", entity.GetComponentTypeName(component), component.GetId())
	}

	return nil
}

func (s *serverImpl) lock(ctx context.Context, key string) (func() error, error) {
	return s.lockWithTimeout(ctx, key, 10)
}

func (s *serverImpl) lockWithTimeout(ctx context.Context, key string, timeoutSecs int) (func() error, error) {
	lock := &entity.Lock{
		Key:         key,
		TimeoutSecs: int32(timeoutSecs),
	}

	unlockFunc, err := s.lockRepo.Lock(ctx, lock)
	if err != nil {
		return nil, err
	}
	return unlockFunc, nil
}

func (s *serverImpl) checkVolumeCanBeModified(ctx context.Context, sharedPool *entity.SharedPool, targetComp *entity.Component) error {
	var needCheckDataVolumeModifyRate bool
	var needCheckRaftLogVolumeModifyRate bool
	for _, comp := range sharedPool.GetComponentGroups() {
		if comp.GetComponentType() != targetComp.GetComponentType() || comp.GetId() != targetComp.GetId() {
			continue
		}
		for _, target := range targetComp.GetStorages() {
			for _, curr := range comp.GetStorages() {
				if target.GetName() != curr.GetName() {
					continue
				}
				if target.GetSizeGi() != 0 && target.GetSizeGi() != curr.GetSizeGi() {
					needCheckDataVolumeModifyRate = target.GetName() == apiv1.Component_Storage_DATA
					needCheckRaftLogVolumeModifyRate = target.GetName() == apiv1.Component_Storage_RAFTLOG
				}
				if target.GetIops() != 0 && target.GetIops() != curr.GetIops() {
					needCheckDataVolumeModifyRate = target.GetName() == apiv1.Component_Storage_DATA
					needCheckRaftLogVolumeModifyRate = target.GetName() == apiv1.Component_Storage_RAFTLOG
				}
				if target.GetThroughputMbs() != 0 && target.GetThroughputMbs() != curr.GetThroughputMbs() {
					needCheckDataVolumeModifyRate = target.GetName() == apiv1.Component_Storage_DATA
					needCheckRaftLogVolumeModifyRate = target.GetName() == apiv1.Component_Storage_RAFTLOG
				}
			}
		}
		break
	}

	if !needCheckDataVolumeModifyRate && !needCheckRaftLogVolumeModifyRate {
		return nil
	}
	var annos map[string]string
	remoteCli, err := s.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		return status.Errorf(codes.Internal, "failed to get remote client for shared pool %s", sharedPool.GetId())
	}
	ns := "tidb" + sharedPool.GetId()
	name := entity.GetComponentGroupName(targetComp)
	switch targetComp.GetComponentType() {
	case apiv1.ComponentType_TIKV:
		tikvgroup := new(v1alpha1.TiKVGroup)
		if err := remoteCli.Get(ctx, client.ObjectKey{
			Name:      name,
			Namespace: ns,
		}, tikvgroup); err != nil {
			tilog.Error(ctx, "failed to get tikv group",
				zap.String("sharedPoolId", sharedPool.GetId()),
				zap.Error(err))
			return status.Errorf(codes.Internal, "failed to get tikv group for shared pool %s", sharedPool.GetId())
		}
		annos = tikvgroup.Annotations
	case apiv1.ComponentType_PD:
		pdgroup := new(v1alpha1.PDGroup)
		if err := remoteCli.Get(ctx, client.ObjectKey{
			Name:      name,
			Namespace: ns,
		}, pdgroup); err != nil {
			tilog.Error(ctx, "failed to get pd group",
				zap.String("sharedPoolId", sharedPool.GetId()),
				zap.Error(err))
			return status.Errorf(codes.Internal, "failed to get pd group for shared pool %s", sharedPool.GetId())
		}
		annos = pdgroup.Annotations
	case apiv1.ComponentType_TIFLASH_WRITE, apiv1.ComponentType_TIFLASH_COMPUTE:
		tiflashgroup := new(v1alpha1.TiFlashGroup)
		if err := remoteCli.Get(ctx, client.ObjectKey{
			Name:      name,
			Namespace: ns,
		}, tiflashgroup); err != nil {
			tilog.Error(ctx, "failed to get tiflash group",
				zap.String("sharedPoolId", sharedPool.GetId()),
				zap.Error(err))
			return status.Errorf(codes.Internal, "failed to get tiflash group for shared pool %s", sharedPool.GetId())
		}
		annos = tiflashgroup.Annotations
	}
	if needCheckDataVolumeModifyRate {
		if val, ok := annos[schemav1.AnnotationsUpdateTimeDataVolumeKey]; ok {
			if ts, err := time.Parse(time.RFC3339, val); err == nil {
				if time.Now().Unix()-ts.Unix() <= 3600*6 {
					return status.Errorf(codes.FailedPrecondition, "storage %s is modified in 6 hours", apiv1.Component_Storage_DATA)
				}
			}
		}
	}
	if needCheckRaftLogVolumeModifyRate {
		if val, ok := annos[schemav1.AnnotationsUpdateTimeRaftLogVolumeKey]; ok {
			if ts, err := time.Parse(time.RFC3339, val); err == nil {
				if time.Now().Unix()-ts.Unix() <= 3600*6 {
					return status.Errorf(codes.FailedPrecondition, "storage %s is modified in 6 hours", apiv1.Component_Storage_RAFTLOG)
				}
			}
		}
	}
	return nil
}

// componentLabelForHPA maps component type to HPA label selector value.
// IMPORTANT: These values MUST match the labels used in HPA templates (e.g., tikv_worker_hpa.templ)
// which use entity.GetComponentTypeName() to set pingcap.com/component label.
func componentLabelForHPA(componentType entity.ComponentType) string {
	switch componentType {
	case apiv1.ComponentType_TIDB:
		return "tidb"
	case apiv1.ComponentType_TIDB_WORKER:
		return entity.TiDBWorkerType // "worker-tidb"
	case apiv1.ComponentType_TIKV_WORKER:
		return entity.TiKVWorkerType // "default-tikv-worker"
	case apiv1.ComponentType_COPROCESSOR_WORKER:
		return entity.CoprocessorWorkerType // "coprocessor-tikv-worker"
	case apiv1.ComponentType_TIPROXY:
		return "tiproxy"
	default:
		return ""
	}
}

func componentSupportsHPA(componentType entity.ComponentType) bool {
	return componentLabelForHPA(componentType) != ""
}

func componentUsesGroupScopedHPA(componentType entity.ComponentType) bool {
	switch componentType {
	case apiv1.ComponentType_TIDB, apiv1.ComponentType_TIPROXY:
		return true
	default:
		return false
	}
}

func ensureCoprocessorWorkerComponentExists(sharedPool *entity.SharedPool) error {
	for _, comp := range sharedPool.GetComponentGroups() {
		if comp == nil {
			continue
		}
		if comp.GetComponentType() == apiv1.ComponentType_COPROCESSOR_WORKER {
			return nil
		}
	}
	return status.Errorf(codes.NotFound, "component %s not found in shared pool %s",
		apiv1.ComponentType_COPROCESSOR_WORKER,
		sharedPool.GetId(),
	)
}

// getHPAForComponent retrieves the HPA for a specific component type in the given namespace.
// For TIDB component type, groupId is required to scope to the correct TiDBGroup's HPA.
// Caller MUST check lo.Contains(entity.EnableHAPComponent, componentType) before calling.
// Returns nil if no HPA exists for the component (not an error condition).
// Returns error only for actual failures during the List operation.
func (s *serverImpl) getHPAForComponent(
	ctx context.Context,
	remoteCli client.Client,
	namespace string,
	componentType entity.ComponentType,
	groupId string,
) (*autoscalingv2.HorizontalPodAutoscaler, error) {
	labelValue := componentLabelForHPA(componentType)

	hpas := new(autoscalingv2.HorizontalPodAutoscalerList)

	labelSelector := map[string]string{
		schemav1.LabelComponentKey: labelValue,
	}
	if componentUsesGroupScopedHPA(componentType) {
		if groupId == "" {
			return nil, status.Errorf(codes.InvalidArgument, "groupId is required for %s component type", componentType)
		}
		labelSelector[schemav1.LabelPingcapGroupKey] = entity.GetComponentGroupName(&entity.Component{
			Id:            groupId,
			ComponentType: componentType,
		})
	}

	if err := remoteCli.List(ctx, hpas, &client.ListOptions{
		Namespace:     namespace,
		LabelSelector: labels.SelectorFromSet(labelSelector),
	}); err != nil {
		return nil, status.Errorf(codes.Internal, "failed to list %s HPAs: %s", labelValue, err)
	}

	if len(hpas.Items) == 0 {
		//nolint:nilnil // No HPA exists, so returning nil HPA and nil error is correct
		return nil, nil
	}

	return &hpas.Items[0], nil
}

// checkHPAScaleDownDisabled checks if HPA scaleDown is disabled for the given component.
// Returns (true, nil) if HPA scaleDown is disabled or component doesn't have HPA.
// Returns (false, nil) if HPA scaleDown is enabled.
// Returns (false, err) if an error occurred during the check.
func (s *serverImpl) checkHPAScaleDownDisabled(
	ctx context.Context,
	sharedPool *entity.SharedPool,
	componentType entity.ComponentType,
	groupId string,
) (bool, error) {
	if !lo.Contains(entity.EnableHAPComponent, componentType) {
		return true, nil
	}

	remoteCli, err := s.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		return false, status.Errorf(codes.Internal, "failed to get remote client for shared pool: %s", err)
	}

	namespace := "tidb" + sharedPool.GetId()
	hpa, err := s.getHPAForComponent(ctx, remoteCli, namespace, componentType, groupId)
	if err != nil {
		return false, err
	}

	if hpa == nil {
		return true, nil
	}

	if hpa.Spec.Behavior == nil || hpa.Spec.Behavior.ScaleDown == nil {
		return false, nil
	}

	if hpa.Spec.Behavior.ScaleDown.SelectPolicy == nil ||
		*hpa.Spec.Behavior.ScaleDown.SelectPolicy != autoscalingv2.DisabledPolicySelect {
		return false, nil
	}

	return true, nil
}

// checkManualScaleHPAGate enforces the HPA gate for manual node-count changes.
// The gate applies only to scale-out, and force=true bypasses it entirely.
func (s *serverImpl) checkManualScaleHPAGate(
	ctx context.Context,
	sharedPool *entity.SharedPool,
	componentType entity.ComponentType,
	groupId string,
	currentNodeCount uint32,
	targetNodeCount uint32,
	force bool,
) error {
	if force || targetNodeCount <= currentNodeCount {
		return nil
	}

	disabled, err := s.checkHPAScaleDownDisabled(ctx, sharedPool, componentType, groupId)
	if err != nil {
		return err
	}
	if !disabled {
		return status.Errorf(codes.FailedPrecondition,
			"cannot scale out %s: HPA scaleDown is enabled. "+
				"Disable it first by calling SetHPAScaleBehavior with scaleDownEnabled=false",
			componentType)
	}

	return nil
}

// findComponentByTypeAndId finds a component in the sharedPool by type and id.
// Returns nil if not found.
func findComponentByTypeAndId(sharedPool *entity.SharedPool, componentType entity.ComponentType, componentId string) *entity.Component {
	for _, comp := range sharedPool.GetComponentGroups() {
		if comp.GetComponentType() == componentType && comp.GetId() == componentId {
			return comp
		}
	}
	return nil
}

func (s *serverImpl) resolveClusterKeyspace(ctx context.Context, clusterId, sharedPoolId string) (string, error) {
	keyspace, keyspaceErr := s.keyspaceFetcher.GetKeyspace(ctx, clusterId, sharedPoolId)
	if keyspaceErr != nil {
		return "", keyspaceErr
	}

	return keyspace, nil
}