package appbox

import (
	"context"
	"encoding/base64"
	"errors"
	"fmt"

	"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"
	"github.com/pingcap/log"
	"github.com/samber/lo"
	clusterctlpb "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/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/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/util"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/clustercli"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/gcpcontainer"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/message"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils"
	rsconfig "github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/appbox"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/approute"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/vpc"
	aksv1beta1 "github.com/upbound/provider-azure/apis/containerservice/v1beta1"
	"go.uber.org/zap"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

type APPBoxService struct {
	cli        client.Client
	vpcr       vpc.VPCReader
	appboxrw   appbox.APPBoxReaderWriter
	approuter  approute.APPRouteReader
	config     rsconfig.RootConfig
	tmplconfig rsconfig.TemplatesConfig
	ccli       *clustercli.ClusterCli
	gkecli     gcpcontainer.ClusterGetter
}

func NewAPPBoxService(
	cli client.Client,
	config rsconfig.RootConfig,
	tmplconfig rsconfig.TemplatesConfig,
) *APPBoxService {
	return &APPBoxService{
		cli:        cli,
		config:     config,
		tmplconfig: tmplconfig,
		vpcr:       vpc.NewVPCReader(cli),
		appboxrw:   appbox.NewAPPBoxReaderWriter(cli),
		approuter:  approute.NewAPPRouteReader(cli),
		ccli:       clustercli.NewClusterCli(config.ClusterControlServiceEndpoint),
	}
}

func (s *APPBoxService) WithGKEClient(gkecli gcpcontainer.ClusterGetter) {
	s.gkecli = gkecli
}

func (s *APPBoxService) CreateAPPBox(ctx context.Context, req *message.CreateAPPBoxReq) (*v1alpha1.APPBox, error) {
	vpcList := &v1alpha1.VPCList{}
	if err := s.vpcr.List(ctx, vpcList, client.MatchingLabels{
		v1alpha1.VPCIDKey: req.VPCID,
	}); err != nil {
		log.Error("failed to list vpc", zap.Error(err))
		return nil, bizerr.New(bizerr.K8SClientError, "failed to list vpc")
	}

	if len(vpcList.Items) == 0 {
		log.Info("vpc not found")
		return nil, bizerr.New(bizerr.ResourceNotFound, "vpc not found")
	}
	if len(vpcList.Items) > 1 {
		log.Info("multiple vpc found")
		return nil, bizerr.New(bizerr.InternalServerError, "multiple vpc found")
	}

	appbox, err := s.ensureAPPBox(ctx, &vpcList.Items[0], req)
	if err != nil {
		return nil, err
	}

	return appbox, nil
}

func (s *APPBoxService) ensureAPPBox(ctx context.Context, vpc *v1alpha1.VPC, req *message.CreateAPPBoxReq) (*v1alpha1.APPBox, error) {
	namer := utils.NewNamer(req.TenantID, req.ProjectID, "", s.config.RegionConfig.Name, s.config.Env)
	version := s.config.APPBoxConfig.GetK8sVersion(namer.GetTenantID(), namer.GetProjectID())
	appbox := &v1alpha1.APPBox{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.Name,
			Namespace: namer.GetNamespace(),
			Labels: map[string]string{
				v1alpha1.RegionKey:         s.config.RegionConfig.Name,
				v1alpha1.VPCNameKey:        vpc.Name,
				v1alpha1.ManageKey:         v1alpha1.ManageFull,
				namer.GetProjectLabelKey(): "true",
				namer.GetTenantLabelKey():  "true",
				// add tags for billing
				v1alpha1.TagProjectKey: namer.GetProjectID(),
				v1alpha1.TagTenantKey:  namer.GetTenantID(),
			},
		},
		Spec: v1alpha1.APPBoxSpec{
			Provider: s.config.Provider,
			Region:   s.config.RegionConfig.Name,
			KubernetesConfig: v1alpha1.KubernetesConfig{
				Version:             version,
				AddonTemplateID:     s.tmplconfig.GetDefaultAddonTemplateId(),
				InfraTemplateID:     s.tmplconfig.GetDefaultInfraTemplateIdWithVersion(s.config.APPBoxConfig.K8sVersion),
				DedicatedTemplateID: s.tmplconfig.GetDefaultDedicatedTemplateIdWithVersion(s.config.APPBoxConfig.K8sVersion),
				DPTemplateID:        s.tmplconfig.GetDefaultDPTemplateIdWithVersion(s.config.APPBoxConfig.K8sVersion),
			},
			VPCRef: &v1alpha1.Reference{
				Name:      vpc.Name,
				Namespace: namer.GetNamespace(),
			},
		},
	}
	appbox.ObjectMeta.Labels = util.MergeMaps(appbox.ObjectMeta.Labels, s.config.ResourceLabels)
	if err := s.appboxrw.Create(ctx, appbox); err != nil {
		if apierrors.IsAlreadyExists(err) {
			log.Info("appbox already exists", zap.String("name", req.Name), zap.String("namespace", req.Namespace))
			return nil, bizerr.New(bizerr.ResourceConflict, "appbox already exists")
		}
		return nil, bizerr.New(bizerr.K8SClientError, "failed to create appbox")
	}
	return appbox, nil
}

func (s *APPBoxService) DeleteAPPBox(ctx context.Context, req *regionalpb.DeleteAPPBoxRequest) error {
	namer := utils.NewNamer("", "", "", s.config.RegionConfig.Name, s.config.Env)
	appbox := &v1alpha1.APPBox{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.Name,
			Namespace: namer.GetNamespace(),
		},
	}
	if err := s.appboxrw.Delete(ctx, appbox); err != nil {
		if apierrors.IsNotFound(err) {
			log.Info("appbox not found", zap.String("name", req.Name))
			return bizerr.New(bizerr.ResourceNotFound, "appbox not found")
		}
		return bizerr.New(bizerr.K8SClientError, "failed to delete appbox")
	}
	return nil
}

func (s *APPBoxService) GetAPPBox(ctx context.Context, req *regionalpb.GetAPPBoxRequest) (*regionalpb.APPBox, error) {
	namer := utils.NewNamer("", "", "", s.config.RegionConfig.Name, s.config.Env)
	appbox := &v1alpha1.APPBox{}
	if err := s.appboxrw.Get(ctx, client.ObjectKey{
		Name:      req.Name,
		Namespace: namer.GetNamespace(),
	}, appbox); err != nil {
		if apierrors.IsNotFound(err) {
			log.Info("appbox not found", zap.String("name", req.Name))
			return nil, bizerr.New(bizerr.ResourceNotFound, "appbox not found")
		}
		return nil, bizerr.New(bizerr.K8SClientError, "failed to get appbox")
	}
	vpc := &v1alpha1.VPC{}
	if appbox.Spec.VPCRef == nil {
		return nil, bizerr.New(bizerr.InternalServerError, "vpc ref is nil")
	}
	if err := s.vpcr.Get(ctx, client.ObjectKey{
		Namespace: appbox.Spec.VPCRef.Namespace,
		Name:      appbox.Spec.VPCRef.Name,
	}, vpc); err != nil {
		return nil, bizerr.New(bizerr.InternalServerError, "failed to get vpc")
	}

	output, err := s.cvtAPPBox2RegionalPB(ctx, appbox, vpc)
	if err != nil {
		log.Error("failed to convert appbox to regionalpb", zap.Error(err))
		return nil, bizerr.New(bizerr.InternalServerError, "failed to convert appbox to regionalpb")
	}
	return output, nil
}

func (s *APPBoxService) ListAPPBoxes(ctx context.Context, req *regionalpb.ListAPPBoxesRequest) (*regionalpb.ListAPPBoxesResponse, error) {
	appboxs := []v1alpha1.APPBox{}
	if req.ClusterId != "" {
		appbox, err := s.getAPPBoxByAPPRout(ctx, req)
		if err != nil {
			return nil, bizerr.New(bizerr.InternalServerError, "failed to get appbox by approute")
		}
		if appbox != nil {
			appboxs = append(appboxs, *appbox)
		}
	} else {
		appboxList, err := s.filterAPPBox(ctx, req)
		if err != nil {
			return nil, bizerr.New(bizerr.InternalServerError, "failed to list appbox")
		}
		appboxs = append(appboxs, appboxList.Items...)
	}
	resp := &regionalpb.ListAPPBoxesResponse{}
	for _, appbox := range appboxs {
		if len(req.Statuses) > 0 {
			if !meetAPPBoxStatus(appbox.Status.Phase, req.Statuses) {
				continue
			}
		}
		vpc := &v1alpha1.VPC{}
		if err := s.vpcr.Get(ctx, client.ObjectKey{
			Namespace: appbox.Spec.VPCRef.Namespace,
			Name:      appbox.Spec.VPCRef.Name,
		}, vpc); err != nil {
			return nil, bizerr.New(bizerr.InternalServerError, "failed to get vpc")
		}

		output, err := s.cvtAPPBox2RegionalPB(ctx, &appbox, vpc)
		if err != nil {
			log.Error("failed to convert appbox to regionalpb", zap.Error(err))
			return nil, bizerr.New(bizerr.InternalServerError, "failed to convert appbox to regionalpb")
		}
		resp.Appboxes = append(resp.Appboxes, output)
	}

	return resp, nil
}

func (s *APPBoxService) cvtAPPBox2RegionalPB(ctx context.Context, appbox *v1alpha1.APPBox, vpc *v1alpha1.VPC) (*regionalpb.APPBox, error) {
	output := &regionalpb.APPBox{
		TenantIds:        utils.GetIDsFromLabel(appbox.GetLabels(), v1alpha1.TenantIDKeyPrefix),
		ProjectIds:       utils.GetIDsFromLabel(appbox.GetLabels(), v1alpha1.ProjectIDKeyPrefix),
		Status:           getAPPBoxStatus(appbox.Status.Phase),
		DataflowAgentUrl: appbox.Status.DataflowAgentUrl,
		ProxyBaseUrl:     appbox.Status.ProxyBaseUrl,
		Name:             appbox.Name,
	}

	switch appbox.Spec.Provider {
	case consts.ProviderTypeEKS:
		kubernetes, err := genEKSKubernets(ctx, s.cli, vpc, appbox)
		if err != nil {
			return nil, err
		}
		// if the appbox was migrated from clusternative arch, we need to get the external id from clusterservice
		if appbox.Labels[v1alpha1.IsLegacyKey] == "true" {
			externalID, err := s.getExternalID(ctx, appbox)
			// if error occuer, do not fail the api, just log the error, otherwise overwrite the external id.
			if err != nil {
				log.Error("failed to get external id", zap.Error(err))
			} else {
				kubernetes.ExternalId = externalID
			}
		}
		output.Kubernetes = kubernetes
	case consts.ProviderTypeGKE:
		kubernetes, err := genGKEKubernetes(ctx, s.cli, s.gkecli, vpc, appbox)
		if err != nil {
			return nil, err
		}
		output.Kubernetes = kubernetes
	case consts.ProviderTypeAKS:
		kubernetes, err := genAKSKubernetes(ctx, s.cli, vpc, appbox, s.config)
		if err != nil {
			return nil, err
		}
		output.Kubernetes = kubernetes
	default:
		return nil, fmt.Errorf("unsupported provider %s", appbox.Spec.Provider)
	}

	return output, nil
}

func genAKSKubernetes(
	ctx context.Context,
	cli client.Client,
	vpc *v1alpha1.VPC,
	appbox *v1alpha1.APPBox,
	config rsconfig.RootConfig,
) (*regionalpb.APPBox_Kubernetes, error) {
	cluster := &aksv1beta1.KubernetesCluster{}
	if err := cli.Get(ctx, client.ObjectKey{Name: appbox.Status.KubernetesClusterName}, cluster); err != nil {
		log.Error("failed to get aks cluster", zap.Error(err))
		return nil, bizerr.New(bizerr.K8SClientError, "failed to get aks cluster")
	}
	if appbox.Status.KubeConfigSecretRef == nil {
		return nil, fmt.Errorf("cannot find kubeconfig secret")
	}
	kubeconfig, err := dataplane.GetDataplaneKubeConfigFromMetaSecret(ctx, cli, appbox.Status.KubeConfigSecretRef)
	if err != nil {
		return nil, fmt.Errorf("failed to get kubeconfig from secret %s/%s, err: %w", appbox.Status.KubeConfigSecretRef.Namespace, appbox.Status.KubeConfigSecretRef.Name, err)
	}
	genSubnets := func(subnets []v1alpha1.Subnet) []*regionalpb.APPBox_Kubernetes_AzureSubnet {
		outputs := make([]*regionalpb.APPBox_Kubernetes_AzureSubnet, 0, len(subnets))
		for _, subnet := range subnets {
			outputs = append(outputs, &regionalpb.APPBox_Kubernetes_AzureSubnet{
				Id: subnet.SubnetID,
			})
		}
		return outputs
	}
	k8s := &regionalpb.APPBox_Kubernetes{
		Name:       cluster.Name,
		Version:    appbox.Spec.KubernetesConfig.Version,
		Endpoint:   fmt.Sprintf("https://%s", ptr.Deref(cluster.Status.AtProvider.Fqdn, "")),
		AccountId:  vpc.Spec.AccountRef.AccountID,
		Vpc:        vpc.Status.VPC.VPCID,
		Kubeconfig: base64.StdEncoding.EncodeToString(kubeconfig),
		Info: &regionalpb.APPBox_Kubernetes_Azure_{
			Azure: &regionalpb.APPBox_Kubernetes_Azure{
				NodeZones:     getZones(vpc, config),
				ResourceGroup: ptr.Deref(cluster.Spec.ForProvider.ResourceGroupName, ""),
				NodeSubnets:   genSubnets(vpc.Status.NodeSubnets),
				PodSubnets:    genSubnets(vpc.Status.PodSubnets),
			},
		},
	}
	return k8s, nil
}

func getZones(vpc *v1alpha1.VPC, conf rsconfig.RootConfig) []string {
	if len(vpc.Spec.Zones) > 0 {
		return vpc.Spec.Zones
	}
	return conf.RegionConfig.Zones
}

func genGKEKubernetes(ctx context.Context, cli client.Client, gkecli gcpcontainer.ClusterGetter, vpc *v1alpha1.VPC, appbox *v1alpha1.APPBox) (*regionalpb.APPBox_Kubernetes, error) {
	cluster := &gkev1beta2.Cluster{}
	if err := cli.Get(ctx, client.ObjectKey{Name: appbox.Status.KubernetesClusterName}, cluster); err != nil {
		log.Error("failed to get gke cluster", zap.Error(err))
		return nil, bizerr.New(bizerr.K8SClientError, "failed to get gke cluster")
	}
	name := fmt.Sprintf("projects/%s/locations/%s/clusters/%s",
		vpc.Spec.AccountRef.AccountID,
		cluster.Status.AtProvider.Location,
		xpmeta.GetExternalName(cluster))
	resp, err := gkecli.GetCluster(context.Background(), &containerpb.GetClusterRequest{
		Name: name,
	})
	if err != nil {
		log.Error("failed to get gke cluster", zap.Error(err))
		return nil, err
	}
	ca := []byte{}
	if resp.MasterAuth != nil {
		ca, err = base64.StdEncoding.DecodeString(resp.MasterAuth.ClusterCaCertificate)
		if err != nil {
			log.Error("failed to decode client certificate", zap.Error(err))
			return nil, err
		}
	}
	k8s := &regionalpb.APPBox_Kubernetes{
		Name:      cluster.Name,
		Version:   appbox.Spec.KubernetesConfig.Version,
		Endpoint:  fmt.Sprintf("https://%s", cluster.Status.AtProvider.Endpoint),
		Ca:        string(ca),
		AccountId: vpc.Spec.AccountRef.AccountID,
		Vpc:       vpc.Status.VPC.VPCID,
		Info: &regionalpb.APPBox_Kubernetes_Gcp_{
			Gcp: &regionalpb.APPBox_Kubernetes_Gcp{
				NodeZones:   cluster.Spec.ForProvider.Locations,
				AdminNodeSa: appbox.Status.IAM.ServiceAccountEmail,
				NodeSubnets: lo.Map(vpc.Status.NodeSubnets, func(subnet v1alpha1.Subnet, _ int) *regionalpb.APPBox_Kubernetes_GcpSubnet {
					return &regionalpb.APPBox_Kubernetes_GcpSubnet{
						Id: subnet.SubnetID,
					}
				}),
			},
		},
	}
	return k8s, nil
}

func genEKSKubernets(ctx context.Context, cli client.Client, vpc *v1alpha1.VPC, appbox *v1alpha1.APPBox) (*regionalpb.APPBox_Kubernetes, error) {
	cluster := &eksv1beta1.Cluster{}
	if err := cli.Get(ctx, client.ObjectKey{Name: appbox.Status.KubernetesClusterName}, cluster); err != nil {
		log.Error("failed to get eks cluster", zap.Error(err))
		return nil, err
	}
	ca, err := base64.StdEncoding.DecodeString(cluster.Status.AtProvider.CertificateAuthorityData)
	if err != nil {
		return nil, err
	}
	genSubnets := func(subnets []v1alpha1.Subnet) []*regionalpb.APPBox_Kubernetes_AwsSubnet {
		outputs := make([]*regionalpb.APPBox_Kubernetes_AwsSubnet, 0, len(subnets))
		for _, subnet := range subnets {
			outputs = append(outputs, &regionalpb.APPBox_Kubernetes_AwsSubnet{
				Id:     subnet.SubnetID,
				AzName: subnet.AvailabilityZone,
				AzId:   subnet.AvailabilityZoneID,
			})
		}
		return outputs
	}
	k8s := &regionalpb.APPBox_Kubernetes{
		Name:      cluster.Name,
		Version:   appbox.Spec.KubernetesConfig.Version,
		Endpoint:  cluster.Status.AtProvider.Endpoint,
		Ca:        string(ca),
		AccountId: vpc.Spec.AccountRef.AccountID,
		// Deprecated: Use GetClusterRole api instead.
		// To keep compatibility, use project id as externalId
		ExternalId: appbox.GetLabels()[v1alpha1.TagProjectKey],
		Vpc:        vpc.Status.VPC.VPCID,
		Info: &regionalpb.APPBox_Kubernetes_Aws_{
			Aws: &regionalpb.APPBox_Kubernetes_Aws{
				SecurityGroup:              cluster.Status.AtProvider.ResourcesVpcConfig.ClusterSecurityGroupID,
				NodeSubnets:                genSubnets(vpc.Status.NodeSubnets),
				PublicLoadbalancerSubnets:  genSubnets(vpc.Status.PublicSubnets),
				PrivateLoadbalancerSubnets: genSubnets(vpc.Status.PrivateSubnets),
				AdminNodeRole:              appbox.Status.IAM.NodeRoleARN,
				TicdcNodeRole:              appbox.Status.TiCDCNodeRoleARN,
				SecondaryNodeSubnets:       genSubnets(vpc.Status.SecondaryNodeSubnets),
			},
		},
	}
	return k8s, nil
}

// getExternalID call cluster-service getClusterRole api to get externalID
func (s *APPBoxService) getExternalID(ctx context.Context, appbox *v1alpha1.APPBox) (string, error) {
	approutes := new(v1alpha1.APPRouteList)
	opt := &client.MatchingLabels{
		v1alpha1.APPBoxNameKey: appbox.Name,
	}
	if err := s.approuter.List(ctx, approutes, client.InNamespace(appbox.Namespace), opt); err != nil {
		log.Error("failed to list approute", zap.String("appbox", client.ObjectKeyFromObject(appbox).String()), zap.Error(err))
		return "", err
	}
	if len(approutes.Items) == 0 {
		log.Error("no approute under appbox", zap.String("appbox", client.ObjectKeyFromObject(appbox).String()))
		return "", errors.New("no approute under appbox")
	}
	in := new(clusterctlpb.GetClusterRoleRequest)
	in.ClusterId = approutes.Items[0].Name
	data, err := s.ccli.GetClusterRole(ctx, in)
	if err != nil {
		log.Error("failed to get cluster role", zap.Any("input", in), zap.Error(err))
		return "", errors.New("no approute under appbox")
	}
	return data.ExternalId, nil
}

func (s *APPBoxService) filterAPPBox(ctx context.Context, req *regionalpb.ListAPPBoxesRequest) (*v1alpha1.APPBoxList, error) {
	appboxList := &v1alpha1.APPBoxList{}
	opt := client.MatchingLabels{
		utils.GenTenantLabelKey(req.TenantId): "true",
	}
	if req.ProjectId != "" {
		opt[utils.GenProjectLabelKey(req.ProjectId)] = "true"
	}
	if err := s.appboxrw.List(ctx, appboxList, opt); err != nil {
		return nil, err
	}
	return appboxList, nil
}

func (s *APPBoxService) getAPPBoxByAPPRout(ctx context.Context, req *regionalpb.ListAPPBoxesRequest) (*v1alpha1.APPBox, error) {
	approuteList := &v1alpha1.APPRouteList{}
	opt := client.MatchingLabels{
		utils.GenTenantLabelKey(req.TenantId):   "true",
		utils.GenClusterLabelKey(req.ClusterId): "true",
	}
	if req.ProjectId != "" {
		opt[utils.GenProjectLabelKey(req.ProjectId)] = "true"
	}
	if err := s.approuter.List(ctx, approuteList, opt); err != nil {
		return nil, err
	}
	if len(approuteList.Items) == 0 {
		return nil, nil
	}
	if len(approuteList.Items) > 1 {
		return nil, fmt.Errorf("multiple approute found")
	}
	targetAPPRoute := approuteList.Items[0]
	appbox := &v1alpha1.APPBox{}
	if err := s.appboxrw.Get(context.Background(), client.ObjectKey{
		Namespace: targetAPPRoute.Spec.APPBoxRef.Namespace,
		Name:      targetAPPRoute.Spec.APPBoxRef.Name,
	}, appbox); err != nil {
		log.Error("failed to get appbox", zap.Error(err))
		return nil, err
	}

	return appbox, nil
}

func meetAPPBoxStatus(phase v1alpha1.APPBoxPhase, statusFilters []regionalpb.APPBox_Status) bool {
	sts := getAPPBoxStatus(phase)
	for _, filter := range statusFilters {
		if sts == filter {
			return true
		}
	}
	return false
}

func getAPPBoxStatus(phase v1alpha1.APPBoxPhase) regionalpb.APPBox_Status {
	switch phase {
	case v1alpha1.APPBoxPhaseCreating, v1alpha1.APPBoxPhasePending, v1alpha1.APPBoxPhaseFailed:
		return regionalpb.APPBox_CREATING
	case v1alpha1.APPBoxPhaseActive:
		return regionalpb.APPBox_ACTIVE
	case v1alpha1.APPBoxPhaseDeleting, v1alpha1.APPBoxPhaseReadyForDelete:
		return regionalpb.APPBox_DELETING
	default:
		return regionalpb.APPBox_STATE_UNSPECIFIED
	}
}