package common

import (
	"context"
	"encoding/base64"
	"fmt"
	"path"
	"time"

	"cloud.google.com/go/container/apiv1/containerpb"
	eksv1beta1 "github.com/crossplane-contrib/provider-aws/apis/eks/v1beta1"
	gkev1beta2 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta2"
	xpmeta "github.com/crossplane/crossplane-runtime/pkg/meta"
	clusterctlpb "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	regionalpb "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	aksv1beta1 "github.com/upbound/provider-azure/apis/containerservice/v1beta1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
	_ "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/constants"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/approute/task"
)

// Cluster call cluster controller to create cluster.
type Cluster struct {
	*task.BaseTask
}

func NewCluster(ctx *task.Context) *Cluster {
	return &Cluster{BaseTask: &task.BaseTask{Context: ctx}}
}

func (t *Cluster) Name() string {
	return "Cluster"
}

func rpbClusterToCcpbCluster(cluster *regionalpb.Cluster) (*clusterctlpb.Cluster, error) {
	ccpbCluster := new(clusterctlpb.Cluster)
	ccpbCluster.Kubernetes = new(clusterctlpb.Kubernetes)
	bs, err := protojson.Marshal(cluster)
	if err != nil {
		return nil, err
	}
	unmarshalOptions := protojson.UnmarshalOptions{
		DiscardUnknown: true,
	}
	if err := unmarshalOptions.Unmarshal(bs, ccpbCluster); err != nil {
		return nil, err
	}
	ccpbCluster.ClusterPhase = clusterctlpb.Cluster_CLUSTER_PHASE_UNSPECIFIED
	return ccpbCluster, nil
}

func (t *Cluster) buildCreateClusterReq(ctx context.Context) (*clusterctlpb.CreateClusterRequest, error) {
	log := t.Logger.WithValues("Cluster", t.APPRoute.Name)
	rpbCluster, err := getClusterFromAPPRoute(t.APPRoute)
	if err != nil {
		log.Error(err, "get cluster from approute failed")
		return nil, err
	}
	ccpbCluster, err := rpbClusterToCcpbCluster(rpbCluster)
	if err != nil {
		log.Error(err, "covert cluster pb failed")
		return nil, err
	}
	ccpbCluster.Kubernetes = new(clusterctlpb.Kubernetes)
	// fill the Kubernetes field
	vpc := new(v1alpha1.VPC)
	{
		key := client.ObjectKey{Namespace: t.APPBox.Spec.VPCRef.Namespace, Name: t.APPBox.Spec.VPCRef.Name}
		// FIXME(shenjun): it's better to use APIReader to get object, doesn't matter for now since other pkg alreay use GET and generate an list-watch goroutine.
		if err := t.Local.Get(ctx, key, vpc); err != nil {
			log.Error(err, "get vpc failed", "vpc", key.String())
			return nil, err
		}
	}
	var vpcID string
	var caData string
	var apiServerEndpoint string
	var k8sClusterName string
	switch t.APPBox.Spec.Provider {
	case string(constants.AWS), consts.ProviderTypeEKS:
		eks := new(eksv1beta1.Cluster)
		key := client.ObjectKey{Name: t.APPBox.Status.KubernetesClusterName}
		if err := t.Local.Get(ctx, key, eks); err != nil {
			log.Error(err, "get eks failed", "eks", key.String())
			return nil, err
		}
		awsParam := new(clusterctlpb.Kubernetes_Aws)
		nodeSubnets := make([]*clusterctlpb.Kubernetes_AwsSubnet, 0)
		secondaryNodeSubnets := make([]*clusterctlpb.Kubernetes_AwsSubnet, 0)
		publicLBSubnets := make([]*clusterctlpb.Kubernetes_AwsSubnet, 0)
		privateLBSubnets := make([]*clusterctlpb.Kubernetes_AwsSubnet, 0)
		for _, subnet := range vpc.Status.NodeSubnets {
			nodeSubnets = append(nodeSubnets, &clusterctlpb.Kubernetes_AwsSubnet{
				Id:     subnet.SubnetID,
				AzId:   subnet.AvailabilityZoneID,
				AzName: subnet.AvailabilityZone,
			})
		}
		for _, subnet := range vpc.Status.PublicSubnets {
			publicLBSubnets = append(publicLBSubnets, &clusterctlpb.Kubernetes_AwsSubnet{
				Id:     subnet.SubnetID,
				AzId:   subnet.AvailabilityZoneID,
				AzName: subnet.AvailabilityZone,
			})
		}
		for _, subnet := range vpc.Status.PrivateSubnets {
			privateLBSubnets = append(privateLBSubnets, &clusterctlpb.Kubernetes_AwsSubnet{
				Id:     subnet.SubnetID,
				AzId:   subnet.AvailabilityZoneID,
				AzName: subnet.AvailabilityZone,
			})
		}
		for _, subnet := range vpc.Status.SecondaryNodeSubnets {
			secondaryNodeSubnets = append(secondaryNodeSubnets, &clusterctlpb.Kubernetes_AwsSubnet{
				Id:     subnet.SubnetID,
				AzId:   subnet.AvailabilityZoneID,
				AzName: subnet.AvailabilityZone,
			})
		}
		k8sClusterName = xpmeta.GetExternalName(eks)
		caData = eks.Status.AtProvider.CertificateAuthorityData
		// This is a defense checking. If endpoint is empty, we have broken the correctness somewhere.
		// The assumption that eks is ready is not valid anymore.
		// Check appbox Status setting logic or approute gardian task logic.
		if eks.Status.AtProvider.Endpoint == "" {
			log.Error(err, "cannot get kubernets endpoint", "name", k8sClusterName)
			return nil, err
		}
		apiServerEndpoint = eks.Status.AtProvider.Endpoint
		vpcID = eks.Status.AtProvider.ResourcesVpcConfig.VpcID
		awsParam.NodeSubnets = nodeSubnets
		awsParam.PublicLoadbalancerSubnets = publicLBSubnets
		awsParam.PrivateLoadbalancerSubnets = privateLBSubnets
		awsParam.SecondaryNodeSubnets = secondaryNodeSubnets
		awsParam.SecurityGroup = eks.Status.AtProvider.ResourcesVpcConfig.ClusterSecurityGroupID
		awsParam.AdminNodeRole = t.APPBox.Status.IAM.NodeRoleARN
		ccpbCluster.Kubernetes.Info = &clusterctlpb.Kubernetes_Aws_{Aws: awsParam}
	case string(constants.GCP), consts.ProviderTypeGKE:
		// protect the process from panic because of nil reference
		if t.GCPContainerCli == nil {
			log.Error(err, "GCPContainerCli in context is nil")
			return nil, err
		}
		gke := new(gkev1beta2.Cluster)
		key := client.ObjectKey{Name: t.APPBox.Status.KubernetesClusterName}
		if err := t.Local.Get(ctx, key, gke); err != nil {
			log.Error(err, "get gke failed", "gke", key.String())
			return nil, err
		}
		name := fmt.Sprintf("projects/%s/locations/%s/clusters/%s",
			vpc.Spec.AccountRef.AccountID,
			gke.Status.AtProvider.Location,
			xpmeta.GetExternalName(gke))

		resp, err := t.GCPContainerCli.GetCluster(ctx, &containerpb.GetClusterRequest{
			Name: name,
		})
		if err != nil {
			log.Error(err, "get cluster from gcp container service failed", "name", name)
			return nil, err
		}
		gcpParam := new(clusterctlpb.Kubernetes_Gcp)
		k8sClusterName = xpmeta.GetExternalName(gke)
		if resp.MasterAuth != nil {
			caData = resp.MasterAuth.ClusterCaCertificate
		}
		// This is a defense checking. If endpoint is empty, we have broken the correctness somewhere.
		// The assumption that gke is ready is not valid anymore.
		// Check appbox Status setting logic or approute gardian task logic.
		if gke.Status.AtProvider.Endpoint != "" {
			apiServerEndpoint = fmt.Sprintf("https://%s", gke.Status.AtProvider.Endpoint)
		} else if resp.Endpoint != "" {
			apiServerEndpoint = fmt.Sprintf("https://%s", resp.Endpoint)
		} else {
			// if we cannot get the endpoint form gke cr or resp, return error to requeue this task.
			log.Error(err, "cannot get kubernets endpoint", "name", name)
			return nil, err

		}
		// TODO(shenjun): full path or network name only???
		// for now, use the full path
		if gke.Status.AtProvider.NetworkConfig != nil {
			vpcID = path.Base(gke.Status.AtProvider.NetworkConfig.Network)
		} else {
			vpcID = vpc.Status.VPC.VPCID
		}
		gcpParam.NodeZones = gke.Spec.ForProvider.Locations
		// TODO(shenjun): get from vpc later, set to empty for now.
		gcpParam.PscSubnet = ""
		gcpParam.AdminNodeSa = t.APPBox.Status.IAM.ServiceAccountEmail
		ccpbCluster.Kubernetes.Info = &clusterctlpb.Kubernetes_Gcp_{Gcp: gcpParam}
	case string(constants.Azure), consts.ProviderTypeAKS:
		aks := &aksv1beta1.KubernetesCluster{}
		if err := t.Local.Get(ctx, client.ObjectKey{Name: t.APPBox.Status.KubernetesClusterName}, aks); err != nil {
			log.Error(err, "get aks failed", "aks", t.APPBox.Status.KubernetesClusterName)
			return nil, err
		}
		apiServerEndpoint = ptr.Deref(aks.Status.AtProvider.Fqdn, "")
		k8sClusterName = xpmeta.GetExternalName(aks)
		azureParam := &clusterctlpb.Kubernetes_Azure{
			NodeZones:         t.getZones(vpc),
			ResourceGroup:     ptr.Deref(aks.Spec.ForProvider.ResourceGroupName, ""),
			NodeResourceGroup: ptr.Deref(aks.Spec.ForProvider.NodeResourceGroup, ""),
		}
		for _, subnet := range vpc.Status.NodeSubnets {
			azureParam.NodeSubnets = append(azureParam.NodeSubnets, &clusterctlpb.Kubernetes_AzureSubnet{
				Id: subnet.SubnetID,
			})
		}
		for _, subnet := range vpc.Status.PodSubnets {
			azureParam.PodSubnets = append(azureParam.PodSubnets, &clusterctlpb.Kubernetes_AzureSubnet{
				Id: subnet.SubnetID,
			})
		}
		ccpbCluster.Kubernetes.Info = &clusterctlpb.Kubernetes_Azure_{Azure: azureParam}
		secretRef := t.APPBox.Status.KubeConfigSecretRef
		kubeconfig, err := dataplane.GetDataplaneKubeConfigFromMetaSecret(ctx, t.Local, secretRef)
		if err != nil {
			return nil, fmt.Errorf("failed to get kubeconfig from secret %s/%s, err: %w", secretRef.Namespace, secretRef.Name, err)
		}
		ccpbCluster.Kubernetes.Kubeconfig = base64.StdEncoding.EncodeToString(kubeconfig)
		vpcID = vpc.Status.VPC.ResourceID
	}
	// fill other k8s info
	ccpbCluster.Kubernetes.Name = k8sClusterName
	ccpbCluster.Kubernetes.Version = t.APPBox.Spec.KubernetesConfig.Version
	ccpbCluster.Kubernetes.AccountId = vpc.Spec.AccountRef.AccountID
	caDataDecoded, err := base64.StdEncoding.DecodeString(caData)
	if err != nil {
		log.Error(err, "base64 decode ca data failed")
		return nil, err
	}
	ccpbCluster.Kubernetes.Ca = string(caDataDecoded)
	ccpbCluster.Kubernetes.Endpoint = apiServerEndpoint
	ccpbCluster.Kubernetes.Vpc = vpcID

	if t.APPRoute.Spec.SharedBucketRef != nil {
		ccpbCluster.SharedBucket = &clusterctlpb.Cluster_SharedBucket{
			BucketUrl: t.buildSharedBucketURLForCluster(rpbCluster.TenantId, rpbCluster.ProjectId, rpbCluster.Id),
		}
	}
	return &clusterctlpb.CreateClusterRequest{Cluster: ccpbCluster}, nil
}

func (t *Cluster) Create(ctx context.Context) deploy.Result {
	if t.APPRoute.Status.Phase == v1alpha1.APPRoutePhaseActive {
		return deploy.Done()
	}
	log := t.Logger.WithValues("Cluster", t.APPRoute.Name)

	// call cluster control service to create
	req, err := t.buildCreateClusterReq(ctx)
	if err != nil {
		log.Error(err, "build create cluster request failed")
		return deploy.Panic(err).WithRequeuAfter(30 * time.Second)
	}
	if _, err := t.ClusterCli.CreateCluster(ctx, req); err != nil {
		if e, ok := status.FromError(err); ok && e.Code() == codes.AlreadyExists {
			t.APPRoute.Status.Phase = v1alpha1.APPRoutePhaseActive
			t.APPRoute.Status.Reason = "cluster already exists"
			return deploy.Done()
		}
		log.Error(err, "create cluster failed")
		return deploy.Panic(err).WithRequeuAfter(30 * time.Second)
	}
	t.APPRoute.Status.Phase = v1alpha1.APPRoutePhaseActive
	t.APPRoute.Status.Reason = "call cluster-service create cluster api successfully"
	return deploy.Done()
}

func (t *Cluster) Delete(ctx context.Context) deploy.Result {
	log := t.Logger.WithValues("Cluster", t.APPRoute.Name)
	//check approute phase
	if t.APPRoute.Status.Phase != v1alpha1.APPRoutePhaseDeleting {
		t.APPRoute.Status.Phase = v1alpha1.APPRoutePhaseDeleting
	}
	// call cluster controller
	// approute cr name is cluster id
	clusterid := t.APPRoute.Name
	if _, err := t.ClusterCli.DeleteCluster(ctx, &clusterctlpb.DeleteClusterRequest{ClusterId: clusterid}); err != nil {
		if e, ok := status.FromError(err); ok {
			if e.Code() == codes.NotFound {
				log.Info("cluster not found, skip delete")
				return deploy.Done()
			}
		}
		log.Error(err, "delete cluster failed")
		return deploy.Panic(err).WithRequeuAfter(30 * time.Second)
	}
	return deploy.Done()
}

func (t *Cluster) PostDelete(ctx context.Context) deploy.Result {
	log := t.Logger.WithValues("Cluster", t.APPRoute.Name)
	// call cluster controller
	// approute cr name is cluster id
	clusterid := t.APPRoute.Name
	_, err := t.ClusterCli.GetCluster(ctx, &clusterctlpb.GetClusterRequest{ClusterId: clusterid})
	if err == nil {
		log.Info("cluster not deleted yet")
		return deploy.Panic(fmt.Errorf("cluster not deleted yet")).WithRequeuAfter(time.Minute)
	}
	if e, ok := status.FromError(err); ok {
		if e.Code() == codes.NotFound {
			log.Info("cluster deleted")
			return deploy.Done()
		}
		log.Error(err, "get cluster failed")
		return deploy.Panic(err).WithRequeuAfter(time.Minute)
	}
	log.Error(err, "get cluster failed")
	return deploy.Panic(err).WithRequeuAfter(time.Minute)
}

func getClusterFromAPPRoute(route *v1alpha1.APPRoute) (*regionalpb.Cluster, error) {
	cluster := new(regionalpb.Cluster)
	// if approute cr contains field which is not defined in regionalpb.Cluster,
	// we consider it's safe to discard the unknown field.
	unmarshalOptions := protojson.UnmarshalOptions{
		DiscardUnknown: true,
	}
	err := unmarshalOptions.Unmarshal([]byte(route.Spec.ForCluster), cluster)
	if err != nil {
		return nil, err
	}
	return cluster, nil
}

func (t *Cluster) buildSharedBucketURLForCluster(tenantId, projectId, clusterId string) string {
	schema := "unknown"
	switch t.APPBox.Spec.Provider {
	case consts.ProviderTypeEKS:
		schema = "s3"
	case consts.ProviderTypeGKE:
		schema = "gcs"
	case consts.ProviderTypeAKS:
		schema = "https"
	}
	return fmt.Sprintf("%s://%s/%s/%s/%s/", schema, t.APPRoute.Spec.SharedBucketRef.BucketName, tenantId, projectId, clusterId)
}

func (t *Cluster) getZones(vpc *v1alpha1.VPC) []string {
	if len(vpc.Spec.Zones) > 0 {
		return vpc.Spec.Zones
	}
	return t.Config.RegionConfig.Zones
}