package service

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/pingcap/tidb-operator/pkg/apis/label"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/samber/lo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	v1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"sigs.k8s.io/controller-runtime/pkg/client"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-agent/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
)

func (s *AgentService) PauseTC(ctx context.Context, req *apiv1.PauseTCReq) (*apiv1.PauseTCResp, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))
	logger.Info("pause tidbcluster", zap.Any("req", req))

	tcs, err := s.getTCsByClusterId(ctx, req.ClusterId)
	if err != nil {
		return nil, err
	}
	var pausedTCNames []string
	for _, tc := range tcs {
		if !req.All && !lo.Contains(req.TcNames, tc.Name) {
			continue
		}

		logger.Info("pause tidbcluster", zap.String("tc", tc.Name))
		tc.Spec.Paused = true
		if err := k8s.Apply(ctx, s.dataplaneCli, tc); err != nil {
			return nil, fmt.Errorf("failed to pause tidbcluster(%s/%s): %w", tc.Namespace, tc.Name, err)
		}
		pausedTCNames = append(pausedTCNames, tc.Name)
	}
	return &apiv1.PauseTCResp{PausedTcNames: pausedTCNames}, nil
}

func (s *AgentService) ResumeTC(ctx context.Context, req *apiv1.ResumeTCReq) (*apiv1.ResumeTCResp, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))
	logger.Info("resume tidbcluster", zap.Any("req", req))

	resumedTCNames, err := s.resumeTC(ctx, &resumeTCParams{
		clusterId: req.ClusterId,
		all:       req.All,
		tcNames:   req.TcNames,
	})
	if err != nil {
		return nil, err
	}
	return &apiv1.ResumeTCResp{ResumedTcNames: resumedTCNames}, nil
}

type resumeTCParams struct {
	clusterId string
	all       bool
	tcNames   []string
}

func (s *AgentService) resumeTC(ctx context.Context, params *resumeTCParams) (resumedTCNames []string, _ error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster_id", params.clusterId))

	tcs, err := s.getTCsByClusterId(ctx, params.clusterId)
	if err != nil {
		return nil, err
	}
	for _, tc := range tcs {
		if !params.all && !lo.Contains(params.tcNames, tc.Name) {
			continue
		}

		logger.Info("resume tidbcluster", zap.String("tc", tc.Name))
		tc.Spec.Paused = false
		// TODO(chenfei): It seems that k8s.Apply can't update Paused from true to false
		if err := s.dataplaneCli.Update(ctx, tc); err != nil {
			return nil, fmt.Errorf("failed to resume tidbcluster(%s/%s): %w", tc.Namespace, tc.Name, err)
		}
		resumedTCNames = append(resumedTCNames, tc.Name)
	}

	return resumedTCNames, nil
}

func (s *AgentService) DeleteTCMember(ctx context.Context, req *apiv1.DeleteTCMemberReq) (*apiv1.DeleteTCMemberResp, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId), zap.String("member_name", req.MemberName))
	logger.Info("delete tc member", zap.Any("req", req))

	// get componentType by pod
	var pod v1.Pod
	if err := s.dataplaneCli.Get(ctx, client.ObjectKey{
		Name:      req.MemberName,
		Namespace: buildNamespace(req.ClusterId),
	}, &pod); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Error(codes.NotFound, err.Error())
		}
		return nil, fmt.Errorf("failed to get pod %s: %w", req.MemberName, err)
	}
	componentTypeStr := pod.Labels[componentLabelKey]
	if _, ok := componentMap[componentTypeStr]; !ok {
		return nil, status.Errorf(codes.InvalidArgument, "pod %s is not a cluster component pod", req.MemberName)
	}
	tcName := pod.Labels[label.InstanceLabelKey]
	if tcName == "" {
		return nil, fmt.Errorf("can't find tc name from pod's annotation, pod=%s", req.MemberName)
	}
	tc, err := s.getTCByClusterId(ctx, req.ClusterId, tcName)
	if err != nil {
		return nil, err
	}
	memberType := v1alpha1.MemberType(componentTypeStr)

	// append this pod to the 'delete-slots' annotation
	annoKeyToDeleteSlot, ok := member2AnnoKeyToDeleteSlot[memberType]
	if !ok {
		return nil, fmt.Errorf("component type %s is not supported to delete slot", memberType)
	}
	annoValue := tc.Annotations[annoKeyToDeleteSlot]
	var slots []int32
	if annoValue != "" {
		if err := json.Unmarshal([]byte(annoValue), &slots); err != nil {
			return nil, fmt.Errorf("unmarshal slots annotation, value=%s: %w", annoValue, err)
		}
	}
	ordinal, err := getOrdinalIdForPod(&pod)
	if err != nil {
		return nil, fmt.Errorf("get pod ordinal id: %w", err)
	}
	if lo.Contains(slots, ordinal) {
		logger.Info("already added to delete slots")
		_, nextOrdinal := calcNextOrdinal(getComponentReplicas(tc, memberType), lo.Without(slots, ordinal))
		newMemberName := fmt.Sprintf("%s%d", pod.GenerateName, nextOrdinal)
		return &apiv1.DeleteTCMemberResp{NewMemberName: newMemberName}, nil
	}
	slots = append(slots, ordinal)
	annoValueBytes, err := json.Marshal(slots)
	if err != nil {
		return nil, fmt.Errorf("marshal slots annotation: %w", err)
	}

	// update tc's annotation
	if tc.Annotations == nil {
		tc.Annotations = map[string]string{}
	}
	tc.Annotations[annoKeyToDeleteSlot] = string(annoValueBytes)
	if err := k8s.Apply(ctx, s.dataplaneCli, tc); err != nil {
		return nil, fmt.Errorf("apply tc: %w", err)
	}

	_, nextOrdinal := calcNextOrdinal(getComponentReplicas(tc, memberType), lo.Without(slots, ordinal))
	newMemberName := fmt.Sprintf("%s%d", pod.GenerateName, nextOrdinal)
	return &apiv1.DeleteTCMemberResp{NewMemberName: newMemberName}, nil
}

func (s *AgentService) ListTCs(ctx context.Context, req *apiv1.ListTCsReq) (*apiv1.ListTCsResp, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))
	logger.Info("list tidbcluster")
	tcs, err := s.getTCsByClusterId(ctx, req.ClusterId)
	if err != nil {
		return nil, err
	}

	out := lo.MapToSlice(tcs, func(k string, v *v1alpha1.TidbCluster) *apiv1.TC {
		return convertTC(v)
	})
	return &apiv1.ListTCsResp{Tcs: out}, nil
}

func convertTC(tc *v1alpha1.TidbCluster) *apiv1.TC {
	out := &apiv1.TC{
		K8SMetadata:      buildK8sMetadata(tc),
		Paused:           tc.Spec.Paused,
		ConfigRevisionId: tc.Annotations[cloudv1.ConfigRevisionIdKey],
	}
	for _, comp := range allManagedComponentsFromTC(tc.AllComponentSpec()) {
		out.ComponentSpecs = append(out.ComponentSpecs, &apiv1.TC_ComponentSpec{
			ComponentType: convertComponentTypeToAPI(comp.MemberType().String()),
			// see cluster-controller/controllers/cluster/task/common/tidbcluster.go:mutateTcForInstanceType
			ProviderInstanceType: comp.Annotations()[consts.TcInstanceTypeAnnotationKey],
			Replicas:             getComponentReplicas(tc, comp.MemberType()),
		})
	}
	return out
}

// allManagedComponentsFromTC filters out components which are not managed by cluster-services, such as ticdc, discovery etc.
func allManagedComponentsFromTC(in []v1alpha1.ComponentAccessor) []v1alpha1.ComponentAccessor {
	var components []v1alpha1.ComponentAccessor
	for _, comp := range in {
		if _, ok := componentMap[comp.MemberType().String()]; !ok {
			continue
		}

		components = append(components, comp)
	}
	return components
}