package proxy

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net/http"
	"strings"

	gkev1beta1 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	"github.com/gin-gonic/gin"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
	"sigs.k8s.io/controller-runtime/pkg/client"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	regionalpb "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/gkeheterogeneous"
	nodepoolutil "github.com/tidbcloud/infra-provider/pkg/nodepool"
)

const gkeHeterogeneousGatekeeperAssignNotReadyReason = "GKE_HETEROGENEOUS_DEPLOY_NOT_READY_FOR_GATEKEEPER"

type regionalGKEHeterogeneousDeployRef struct {
	componentType    regionalpb.ComponentType
	componentName    cloudv1.ComponentType
	tidbGroupID      string
	computeClassName string
	bindings         []cloudv1.GKEComputeClassNodePoolBinding
}

func dropTiDBGroupQueryForTiKVGatekeeperAssign(c *gin.Context) {
	if c.Request.Method != http.MethodGet || !strings.HasSuffix(c.FullPath(), "/gkeHeterogeneous/gatekeeperAssigns/:componentType") {
		return
	}
	componentType, ok := regionalGKEHeterogeneousComponentTypeFromPath(c.Param("componentType"))
	if !ok || componentType != regionalpb.ComponentType_TIKV {
		return
	}
	query := c.Request.URL.Query()
	query.Del("tidb_group_id")
	query.Del("tidbGroupId")
	c.Request.URL.RawQuery = query.Encode()
}

func regionalGKEHeterogeneousComponentTypeFromPath(componentType string) (regionalpb.ComponentType, bool) {
	switch strings.ToUpper(strings.TrimSpace(componentType)) {
	case regionalpb.ComponentType_TIKV.String():
		return regionalpb.ComponentType_TIKV, true
	case regionalpb.ComponentType_TIDB.String():
		return regionalpb.ComponentType_TIDB, true
	default:
		return regionalpb.ComponentType_COMPONENT_TYPE_UNSPECIFIED, false
	}
}

func (p *ClusterAgentProxy) rejectGKEHeterogeneousGatekeeperAssignNotReady(c *gin.Context, cluster *cloudv1.Cluster) bool {
	rejectIfNotReady := func(computeClassName string, ref regionalGKEHeterogeneousDeployRef, refErr error) bool {
		if refErr != nil {
			st := status.Convert(refErr)
			abortWithGRPCError(c, st.Code(), st.Message(), gkeHeterogeneousGatekeeperAssignNotReadyReason)
			return true
		}
		if err := validateGatekeeperAssignComputeClassName(computeClassName, ref); err != nil {
			st := status.Convert(err)
			abortWithGRPCError(c, st.Code(), st.Message(), "")
			return true
		}
		if err := p.ensureGKEHeterogeneousDeployReadyForGatekeeper(c.Request.Context(), cluster, ref); err != nil {
			st := status.Convert(err)
			abortWithGRPCError(c, st.Code(), st.Message(), gkeHeterogeneousGatekeeperAssignNotReadyReason)
			return true
		}
		return false
	}

	switch {
	case c.Request.Method == http.MethodPost && strings.HasSuffix(c.FullPath(), "/gkeHeterogeneous/gatekeeperAssigns"):
		req := &regionalpb.CreateGKEHeterogeneousGatekeeperAssignRequest{}
		if err := readProtoJSONBody(c, req); err != nil {
			abortWithGRPCError(c, codes.InvalidArgument, err.Error(), "")
			return true
		}
		ref, err := regionalGKEHeterogeneousDeployRefForRequest(cluster, req.GetComponentType(), req.GetTidbGroupId())
		return rejectIfNotReady(req.GetComputeClassName(), ref, err)
	case c.Request.Method == http.MethodPatch && strings.HasSuffix(c.FullPath(), "/gkeHeterogeneous/gatekeeperAssigns/:name"):
		req := &regionalpb.UpdateGKEHeterogeneousGatekeeperAssignRequest{}
		if err := readProtoJSONBody(c, req); err != nil {
			abortWithGRPCError(c, codes.InvalidArgument, err.Error(), "")
			return true
		}
		ref, err := regionalGKEHeterogeneousDeployRefForAssignName(cluster, c.Param("name"))
		return rejectIfNotReady(req.GetComputeClassName(), ref, err)
	default:
		return false
	}
}

func readProtoJSONBody(c *gin.Context, msg proto.Message) error {
	body, err := io.ReadAll(c.Request.Body)
	if err != nil {
		return fmt.Errorf("read request body: %s", err)
	}
	c.Request.Body = io.NopCloser(bytes.NewReader(body))
	if len(bytes.TrimSpace(body)) == 0 {
		return fmt.Errorf("request body is required")
	}
	if err := (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(body, msg); err != nil {
		return fmt.Errorf("decode request body: %s", err)
	}
	c.Request.Body = io.NopCloser(bytes.NewReader(body))
	return nil
}

func regionalGKEHeterogeneousDeployRefForAssignName(cluster *cloudv1.Cluster, assignName string) (regionalGKEHeterogeneousDeployRef, error) {
	assignName = strings.TrimSpace(assignName)
	if assignName == "" {
		return regionalGKEHeterogeneousDeployRef{}, status.Error(codes.InvalidArgument, "gatekeeper assign name is required")
	}
	for _, ref := range activeRegionalGKEHeterogeneousDeployRefs(cluster) {
		expectedName := gkeheterogeneous.ResourceName(cluster.Name, ref.componentName, ref.tidbGroupID, "compute-class-assign")
		if expectedName == assignName {
			return ref, nil
		}
	}
	return regionalGKEHeterogeneousDeployRef{}, status.Errorf(codes.FailedPrecondition, "gatekeeper assign %s does not match any active heterogeneous deploy intent", assignName)
}

func regionalGKEHeterogeneousDeployRefForRequest(cluster *cloudv1.Cluster, componentType regionalpb.ComponentType, tidbGroupID string) (regionalGKEHeterogeneousDeployRef, error) {
	tidbGroupID = strings.TrimSpace(tidbGroupID)
	if componentType != regionalpb.ComponentType_TIDB {
		// tidb_group_id only selects TiDB groups. TiKV gatekeeper assign readiness
		// uses one intent per cluster/component, so request group values are ignored.
		tidbGroupID = ""
	}
	for _, ref := range activeRegionalGKEHeterogeneousDeployRefs(cluster) {
		if ref.componentType == componentType && strings.TrimSpace(ref.tidbGroupID) == tidbGroupID {
			return ref, nil
		}
	}
	return regionalGKEHeterogeneousDeployRef{}, status.Errorf(codes.FailedPrecondition, "heterogeneous deploy for %s is not ready to create gatekeeper assign", componentType)
}

func activeRegionalGKEHeterogeneousDeployRefs(cluster *cloudv1.Cluster) []regionalGKEHeterogeneousDeployRef {
	if cluster == nil || cluster.GetProviderType() != cloudv1.ProviderTypeGKE {
		return nil
	}
	out := make([]regionalGKEHeterogeneousDeployRef, 0)
	if ref, ok := decodeRegionalGKEHeterogeneousDeployRef(cluster, regionalpb.ComponentType_TIKV, "", cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, cluster.Spec.Cluster.TiKV.InstanceType); ok {
		out = append(out, ref)
	}
	for _, groupID := range cluster.ActiveTiDBGKEComputeClassGroupIDs() {
		ref, ok := decodeRegionalGKEHeterogeneousDeployRef(cluster, regionalpb.ComponentType_TIDB, groupID, cluster.TiDBGKEComputeClassBindingsAnnoKey(groupID), cluster.TiDBInstanceTypeForGroup(groupID))
		if ok {
			out = append(out, ref)
		}
	}
	return out
}

func decodeRegionalGKEHeterogeneousDeployRef(cluster *cloudv1.Cluster, componentType regionalpb.ComponentType, tidbGroupID, annotationKey, baseInstanceType string) (regionalGKEHeterogeneousDeployRef, bool) {
	if strings.TrimSpace(cluster.GetAnnotations()[annotationKey]) == "" {
		return regionalGKEHeterogeneousDeployRef{}, false
	}
	baseInstanceType = strings.TrimSpace(baseInstanceType)
	if baseInstanceType == "" {
		return regionalGKEHeterogeneousDeployRef{}, false
	}
	annotation, explicit, err := cloudv1.ClusterGKEComputeClassBindingAnnotation(
		cluster,
		annotationKey,
		baseInstanceType,
		gkeheterogeneous.NodePoolName(cluster, baseInstanceType),
	)
	if err != nil || !explicit {
		return regionalGKEHeterogeneousDeployRef{}, false
	}
	componentName := cloudv1.ComponentTiKV
	if componentType == regionalpb.ComponentType_TIDB {
		componentName = cloudv1.ComponentTiDB
	}
	return regionalGKEHeterogeneousDeployRef{
		componentType:    componentType,
		componentName:    componentName,
		tidbGroupID:      strings.TrimSpace(tidbGroupID),
		computeClassName: strings.TrimSpace(annotation.ComputeClassName),
		bindings:         annotation.NodePools,
	}, true
}

func validateGatekeeperAssignComputeClassName(requested string, ref regionalGKEHeterogeneousDeployRef) error {
	normalized, err := cloudv1.NormalizeGKEComputeClassName(requested)
	if err != nil {
		return status.Error(codes.InvalidArgument, err.Error())
	}
	if normalized != strings.TrimSpace(ref.computeClassName) {
		return status.Errorf(codes.InvalidArgument, "compute_class_name %s does not match heterogeneous deploy compute class %s", normalized, ref.computeClassName)
	}
	return nil
}

func (p *ClusterAgentProxy) ensureGKEHeterogeneousDeployReadyForGatekeeper(ctx context.Context, cluster *cloudv1.Cluster, ref regionalGKEHeterogeneousDeployRef) error {
	if p == nil || p.kubeClient == nil {
		return status.Error(codes.Internal, "cluster agent proxy kube client is not configured")
	}
	for _, binding := range ref.bindings {
		nodePool := &gkev1beta1.NodePool{}
		if err := p.kubeClient.Get(ctx, client.ObjectKey{Name: binding.NodePoolName}, nodePool); err != nil {
			return status.Errorf(codes.FailedPrecondition, "get heterogeneous nodepool %s: %s", binding.NodePoolName, err)
		}
		if err := validateRegionalGKEHeterogeneousNodePoolBinding(cluster, nodePool, binding); err != nil {
			return err
		}
		ready, err := gkeheterogeneous.NodePoolReadyForGatekeeper(nodePool, ref.computeClassName)
		if err != nil {
			return err
		}
		if !ready {
			return status.Errorf(codes.FailedPrecondition, "heterogeneous nodepool %s is not ready for gatekeeper assign", binding.NodePoolName)
		}
	}
	return nil
}

func validateRegionalGKEHeterogeneousNodePoolBinding(cluster *cloudv1.Cluster, nodePool *gkev1beta1.NodePool, binding cloudv1.GKEComputeClassNodePoolBinding) error {
	if nodePool == nil {
		return status.Error(codes.Internal, "heterogeneous nodepool is nil")
	}
	if nodePool.GetLabels()[consts.GKEHeterogeneousNodePoolManagedByLabelKey] != consts.GKEHeterogeneousNodePoolManagedByLabelValue {
		return status.Errorf(codes.FailedPrecondition, "heterogeneous nodepool %s is not managed by heterogeneous nodepool API", nodePool.GetName())
	}
	if nodePool.Spec.ForProvider.Config == nil || nodePool.Spec.ForProvider.Config.MachineType == nil {
		return status.Errorf(codes.FailedPrecondition, "heterogeneous nodepool %s does not have a machine type configured", nodePool.GetName())
	}
	if actual := strings.TrimSpace(*nodePool.Spec.ForProvider.Config.MachineType); actual != strings.TrimSpace(binding.InstanceType) {
		return status.Errorf(codes.FailedPrecondition, "heterogeneous nodepool %s has machine type %s, want %s", nodePool.GetName(), actual, binding.InstanceType)
	}
	scopeLabels, missingScopeLabels := nodepoolutil.GKEHeterogeneousNodePoolScopeLabels(cluster)
	if len(missingScopeLabels) > 0 {
		return status.Errorf(codes.FailedPrecondition, "cluster %s/%s is missing required GKE heterogeneous nodepool scope labels: %s", cluster.Namespace, cluster.Name, strings.Join(missingScopeLabels, ","))
	}
	if err := gkeheterogeneous.ValidateNodePoolScopeLabels(nodePool, cluster.GetName(), "GKE heterogeneous nodepool scope", scopeLabels); err != nil {
		return status.Error(codes.FailedPrecondition, err.Error())
	}
	expectedClusterRef := strings.TrimSpace(regionalGKEHeterogeneousNodePoolClusterRef(cluster))
	if actual := strings.TrimSpace(nodePool.Spec.ForProvider.Cluster); actual != expectedClusterRef {
		return status.Errorf(codes.PermissionDenied, "heterogeneous nodepool %s does not belong to cluster %s GKE cluster ref: cluster=%q, want %q", nodePool.GetName(), cluster.GetName(), actual, expectedClusterRef)
	}
	return nil
}

func regionalGKEHeterogeneousNodePoolClusterRef(cluster *cloudv1.Cluster) string {
	if cluster == nil {
		return ""
	}
	return fmt.Sprintf("projects/%s/locations/%s/clusters/%s", cluster.Spec.Kubernetes.Account, strings.TrimSpace(cluster.GetRegion()), cluster.Spec.Kubernetes.Name)
}