package tioperator

import (
	"context"
	"fmt"
	"maps"
	"strconv"
	"strings"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	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"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	apimeta "k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	placementPolicyAPIVersion       = "core.pingcap.com/v1alpha1"
	placementPolicyKind             = "PlacementPolicy"
	placementPolicyGroup            = "core.pingcap.com"
	placementPolicyTiKVGroupKind    = "TiKVGroup"
	placementPolicyRuleName         = "logical-cluster"
	placementPolicyLabelTiKVGroup   = "tags.tidbcloud.com/tikvgroup"
	placementPolicyVoterCount       = int32(3)
	placementPolicyComponentCluster = clusterCRName
)

func ListPlacementPolicies(ctx context.Context, remoteCli client.Client, sharedPoolID, tikvGroupID string) ([]*entity.PlacementPolicy, error) {
	var policyList v1alpha1.PlacementPolicyList
	listOptions := []client.ListOption{client.InNamespace(dataplaneNamespace(sharedPoolID))}
	if tikvGroupID != "" {
		listOptions = append(listOptions, client.MatchingLabels{
			placementPolicyLabelTiKVGroup: placementPolicyTiKVGroupName(tikvGroupID),
		})
	}
	if err := remoteCli.List(ctx, &policyList, listOptions...); err != nil {
		return nil, status.Errorf(codes.Internal, "failed to list placement policies: %s", err)
	}

	policies := make([]*entity.PlacementPolicy, 0, len(policyList.Items))
	for i := range policyList.Items {
		policyCR := &policyList.Items[i]
		policy, err := cvtPlacementPolicyToEn(policyCR)
		if err != nil {
			tilog.Warn(ctx, "convert placement policy to entity", zap.String("name", policyCR.Name), zap.Error(err))
			continue
		}
		policy.SharedPoolId = sharedPoolID
		policies = append(policies, policy)
	}

	return policies, nil
}

func GetPlacementPolicy(ctx context.Context, remoteCli client.Client, sharedPoolID, placementPolicyID string) (*entity.PlacementPolicy, error) {
	policyCR, err := getPlacementPolicyCR(ctx, remoteCli, dataplaneNamespace(sharedPoolID), placementPolicyID)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "placement policy %s not found: %s", placementPolicyID, err)
		}
		return nil, status.Errorf(codes.Internal, "failed to get placement policy %s: %s", placementPolicyID, err)
	}

	policy, err := cvtPlacementPolicyToEn(policyCR)
	if err != nil {
		return nil, err
	}
	policy.SharedPoolId = sharedPoolID
	return policy, nil
}

func CreatePlacementPolicy(ctx context.Context, remoteCli client.Client, sharedPoolID string, policy *entity.PlacementPolicy, keyspaceID uint32) (*entity.PlacementPolicy, error) {
	placementPolicyID := policy.GetPlacementPolicyId()
	clusterID := policy.GetClusterId()
	expected := buildPlacementPolicyCR(sharedPoolID, placementPolicyID, clusterID, policy.GetTikvGroupId(), keyspaceID)
	if err := remoteCli.Create(ctx, expected); err != nil {
		if apierrors.IsAlreadyExists(err) {
			return nil, status.Errorf(codes.AlreadyExists, "placement policy %s already exists", placementPolicyID)
		}
		return nil, status.Errorf(codes.Internal, "failed to create placement policy %s: %s", placementPolicyID, err)
	}

	out, err := cvtPlacementPolicyToEn(expected)
	if err != nil {
		return nil, err
	}
	out.SharedPoolId = sharedPoolID
	return out, nil
}

func UpdatePlacementPolicy(ctx context.Context, remoteCli client.Client, sharedPoolID string, policy *entity.PlacementPolicy, keyspaceID uint32) (*entity.PlacementPolicy, error) {
	placementPolicyID := policy.GetPlacementPolicyId()
	clusterID := policy.GetClusterId()
	namespace := dataplaneNamespace(sharedPoolID)
	expected := buildPlacementPolicyCR(sharedPoolID, placementPolicyID, clusterID, policy.GetTikvGroupId(), keyspaceID)

	current, err := getPlacementPolicyCR(ctx, remoteCli, namespace, placementPolicyID)
	if apierrors.IsNotFound(err) {
		return nil, status.Errorf(codes.NotFound, "placement policy %s not found: %s", placementPolicyID, err)
	}
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get placement policy %s: %s", placementPolicyID, err)
	}

	current.Labels = updatePlacementPolicyLabels(current.Labels, expected.Labels)
	current.Spec = expected.Spec
	if err := remoteCli.Update(ctx, current); err != nil {
		return nil, status.Errorf(codes.Internal, "failed to update placement policy %s: %s", placementPolicyID, err)
	}

	out, err := cvtPlacementPolicyToEn(current)
	if err != nil {
		return nil, err
	}
	out.SharedPoolId = sharedPoolID
	return out, nil
}

func DeletePlacementPolicy(ctx context.Context, remoteCli client.Client, sharedPoolID, placementPolicyID string) error {
	if err := remoteCli.Delete(ctx, &v1alpha1.PlacementPolicy{
		ObjectMeta: metav1.ObjectMeta{
			Name: placementPolicyID, Namespace: dataplaneNamespace(sharedPoolID),
		},
	}); err != nil {
		// Dataplanes without the PlacementPolicy API have nothing to delete.
		if apierrors.IsNotFound(err) || apimeta.IsNoMatchError(err) {
			return status.Errorf(codes.NotFound, "placement policy %s not found: %s", placementPolicyID, err)
		}
		return status.Errorf(codes.Internal, "failed to delete placement policy %s: %s", placementPolicyID, err)
	}
	return nil
}

func getPlacementPolicyCR(ctx context.Context, remoteCli client.Client, dataplaneNs, placementPolicyID string) (*v1alpha1.PlacementPolicy, error) {
	var policy v1alpha1.PlacementPolicy
	key := types.NamespacedName{Name: placementPolicyID, Namespace: dataplaneNs}
	if err := remoteCli.Get(ctx, key, &policy); err != nil {
		return nil, err
	}
	return &policy, nil
}

func buildPlacementPolicyCR(sharedPoolID, placementPolicyID, clusterID, tikvGroupID string, keyspaceID uint32) *v1alpha1.PlacementPolicy {
	keyspaceIDString := strconv.FormatUint(uint64(keyspaceID), 10)
	tikvGroupName := placementPolicyTiKVGroupName(tikvGroupID)

	return &v1alpha1.PlacementPolicy{
		ObjectMeta: metav1.ObjectMeta{
			Name:      placementPolicyID,
			Namespace: dataplaneNamespace(sharedPoolID),
			Labels: map[string]string{
				schemav1.LabelSharedPoolKey:   sharedPoolID,
				schemav1.LabelUserClusterKey:  clusterID,
				placementPolicyLabelTiKVGroup: tikvGroupName,
			},
		},
		Spec: v1alpha1.PlacementPolicySpec{
			Cluster: v1alpha1.ClusterReference{
				Name: placementPolicyComponentCluster,
			},
			GroupRefs: []v1alpha1.PlacementPolicyGroupRef{
				{
					Group: placementPolicyGroup,
					Kind:  placementPolicyTiKVGroupKind,
					Name:  tikvGroupName,
				},
			},
			Rules: []v1alpha1.PlacementPolicyRule{
				{
					Name:  placementPolicyRuleName,
					Role:  v1alpha1.PlacementPolicyRoleVoter,
					Count: placementPolicyVoterCount,
					Selector: v1alpha1.PlacementPolicySelector{
						Keyspace: &v1alpha1.PlacementPolicyKeyspaceSelector{
							IDs: []string{keyspaceIDString},
						},
					},
				},
			},
		},
	}
}

func cvtPlacementPolicyToEn(policyCR *v1alpha1.PlacementPolicy) (*entity.PlacementPolicy, error) {
	if len(policyCR.Spec.GroupRefs) != 1 {
		return nil, status.Errorf(codes.FailedPrecondition, "placement policy %s must have exactly one group ref", policyCR.Name)
	}
	if len(policyCR.Spec.Rules) != 1 {
		return nil, status.Errorf(codes.FailedPrecondition, "placement policy %s must have exactly one rule", policyCR.Name)
	}

	groupRef := policyCR.Spec.GroupRefs[0]
	if groupRef.Group != placementPolicyGroup || groupRef.Kind != placementPolicyTiKVGroupKind {
		return nil, status.Errorf(codes.FailedPrecondition, "placement policy %s has unsupported group ref %s/%s", policyCR.Name, groupRef.Group, groupRef.Kind)
	}
	tikvGroupID, err := tikvGroupIDFromPlacementGroupRef(groupRef.Name)
	if err != nil {
		return nil, status.Errorf(codes.FailedPrecondition, "placement policy %s has invalid group ref name: %s", policyCR.Name, err)
	}
	clusterID := policyCR.Labels[schemav1.LabelUserClusterKey]
	if clusterID == "" {
		clusterID = policyCR.Name
	}
	if clusterID != policyCR.Name {
		return nil, status.Errorf(codes.FailedPrecondition, "placement policy %s cluster id %s must match placement policy id", policyCR.Name, clusterID)
	}

	out := &entity.PlacementPolicy{
		PlacementPolicyId: policyCR.Name,
		ClusterId:         clusterID,
		TikvGroupId:       tikvGroupID,
		Phase:             getPlacementPolicyPhase(policyCR),
	}
	return out, nil
}

func getPlacementPolicyPhase(policyCR *v1alpha1.PlacementPolicy) apiv1.PlacementPolicy_Phase {
	if len(policyCR.Status.Conditions) == 0 {
		return apiv1.PlacementPolicy_CREATING
	}
	if !isSynced(policyCR.Status.Conditions) {
		return apiv1.PlacementPolicy_MODIFYING
	}
	if policyCR.Status.ObservedGeneration != 0 && policyCR.GetGeneration() > policyCR.Status.ObservedGeneration {
		return apiv1.PlacementPolicy_MODIFYING
	}
	return apiv1.PlacementPolicy_AVAILABLE
}

func updatePlacementPolicyLabels(existing, expected map[string]string) map[string]string {
	labels := existing
	if labels == nil {
		labels = map[string]string{}
	}
	maps.Copy(labels, expected)
	return labels
}

func tikvGroupIDFromPlacementGroupRef(name string) (string, error) {
	if !strings.HasPrefix(name, componentGroupPrefix) {
		return "", fmt.Errorf("%s must have %s prefix", name, componentGroupPrefix)
	}
	return strings.TrimPrefix(name, componentGroupPrefix), nil
}

func placementPolicyTiKVGroupName(tikvGroupID string) string {
	return fmt.Sprintf("%s%s", componentGroupPrefix, tikvGroupID)
}