package vpcpeering

import (
	"context"
	"errors"
	"fmt"

	"github.com/pingcap/log"
	regionalapipb "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"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/constants"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"go.uber.org/zap"
	emptypb "google.golang.org/protobuf/types/known/emptypb"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/selection"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
)

type Service struct {
	// we have some special case when creating VPCPeering CR, need to read kube-apiserver directly.
	kubeClient    client.Client // access kube-apiserver with mgr cache.
	kubeAPIReader client.Reader // access kube-apiserver directly.

	config   config.RootConfig
	provider string // value can be `eks` or `gke`
	region   string // phyical region, value can be `us-west-2`, `us-east-2`, `us-west1`
}

type Option func(*Service) error

var providerSupported = map[string]string{
	"aws":   "aws",
	"gcp":   "gcp",
	"azure": "azure",
	"eks":   "aws",
	"gke":   "gcp",
	"aks":   "aks",
}

func WithProvider(provider string) Option {
	return func(svc *Service) error {
		if v, ok := providerSupported[provider]; !ok {
			return fmt.Errorf("provider %s not support", provider)
		} else {
			svc.provider = v
		}
		return nil
	}
}

func WithRegion(region string) Option {
	return func(svc *Service) error {
		svc.region = region
		return nil
	}
}

func WithConfig(config config.RootConfig) Option {
	return func(svc *Service) error {
		svc.config = config
		return nil
	}
}

func NewService(kubeClient client.Client, kubeAPIReader client.Reader, opts ...Option) (*Service, error) {
	svc := &Service{
		kubeClient:    kubeClient,
		kubeAPIReader: kubeAPIReader,
	}
	for _, opt := range opts {
		if err := opt(svc); err != nil {
			return nil, err
		}
	}
	return svc, nil
}

func (svc *Service) GetProvider() string {
	return svc.provider
}

// Create
// @Description: create a new vpcpeering
// @Parameter ctx
// @Parameter req message.CreateVPCPeeringResp
// @return message.CreateVPCPeeringResp
// @return errors.Error
func (svc *Service) Create(ctx context.Context, req *regionalapipb.CreateVpcPeeringRequest) (*regionalapipb.VpcPeering, error) {
	vpcID := ""
	var vpc *v1alpha1.VPC

	switch svc.provider {
	case string(constants.AWS):
		vpcID = req.GetVpcPeering().GetAws().GetSourceVpcId()
	case string(constants.GCP):
		vpcID = req.GetVpcPeering().GetGcp().GetSourceNetworkName()
	default:
		return nil, bizerr.New(bizerr.ConfigInvalid, "unknown vendor")
	}
	// we must use label instead of CR name to find the source VPC
	{
		list := new(v1alpha1.VPCList)
		tenantIDLabelKey := utils.GenTenantLabelKey(req.VpcPeering.GetTenantId())
		projectIDLabelKey := utils.GenProjectLabelKey(req.VpcPeering.GetProjectId())
		labelselector := client.MatchingLabels{
			v1alpha1.VPCIDKey: vpcID,
			tenantIDLabelKey:  "true",
			projectIDLabelKey: "true",
		}
		if err := svc.kubeClient.List(ctx, list, labelselector); err != nil {
			if apierrors.IsNotFound(err) {
				return nil, bizerr.New(bizerr.ParameterInvalid, "souce vpc cr not existed")
			}
			log.Error("get vpc cr failed", zap.String("vpcID", vpcID), zap.Error(err))
			return nil, bizerr.New(bizerr.K8SClientError, err.Error())
		}
		if len(list.Items) == 0 {
			return nil, bizerr.New(bizerr.ParameterInvalid, "souce vpc cr not existed")
		}
		vpc = &(list.Items[0])
	}

	vpcpeering := new(v1alpha1.VPCPeering)
	if err := svc.fillVPCPeeringCR(vpcpeering, req.GetVpcPeering(), vpc); err != nil {
		log.Error("fill vpcpeering failed", zap.Error(err))
		return nil, bizerr.New(bizerr.ParameterInvalid, err.Error())
	}
	// TODO(shenjun): ask global team about the abort on existed behaviour
	if err := svc.kubeClient.Create(ctx, vpcpeering); err != nil {
		log.Error("create vpcpeering failed", zap.Error(err), zap.String("namespace", vpcpeering.Namespace), zap.String("name", vpcpeering.Name))
		if !apierrors.IsAlreadyExists(err) {
			return nil, bizerr.New(bizerr.K8SClientError, err.Error())
		}
	}
	vpcpeeringpb := new(regionalapipb.VpcPeering)
	if err := fillVPCPeeringPB(vpcpeeringpb, vpcpeering); err != nil {
		log.Error("fille vpcpeering pb failed", zap.Error(err), zap.String("vpcpeering", client.ObjectKeyFromObject(vpcpeering).String()))
		return nil, bizerr.New(bizerr.K8SClientError, err.Error())
	}
	return vpcpeeringpb, nil
}

// Delete
// @Description: delete a existed vpcpeering
// @Parameter ctx
// @Parameter vpcPeeringId string
// @return message.DeleteVPCPeeringResp
// @return errors.Error
func (svc *Service) Delete(ctx context.Context, req *regionalapipb.DeleteVpcPeeringRequest) (*emptypb.Empty, error) {
	// NOTE(shenjun): VpcPeeringId in delete request is same as Id in create request
	id := req.GetVpcPeeringId()
	resp := new(emptypb.Empty)
	vpcpeerings := v1alpha1.VPCPeeringList{}
	err := svc.kubeClient.List(ctx, &vpcpeerings, client.MatchingLabels{
		v1alpha1.LabelKeyBizID: id,
	})
	if client.IgnoreNotFound(err) != nil {
		log.Error("list vpcpeering cr failed", zap.String("id", id), zap.Error(err))
		return nil, bizerr.New(bizerr.K8SClientError, err.Error())
	}
	if apierrors.IsNotFound(err) || len(vpcpeerings.Items) == 0 {
		log.Info("vpcpeering cr not found", zap.String("id", id))
		return nil, bizerr.New(bizerr.ResourceNotFound, "vpcpeering cr not found")
	}
	vpcpeering := vpcpeerings.Items[0]
	err = svc.kubeClient.Delete(ctx, &vpcpeering)
	if client.IgnoreNotFound(err) != nil {
		log.Error("delete vpcpeering cr failed", zap.String("namespace", vpcpeering.GetNamespace()), zap.String("name", vpcpeering.GetName()), zap.Error(err))
		return nil, bizerr.New(bizerr.K8SClientError, err.Error())
	}
	return resp, nil
}

// Get
// @Description: get a existed vpcpeering
// @Parameter ctx
// @Parameter req message.CreateVPCPeeringResp
// @return message.CreateVPCPeeringResp
// @return errors.Error
func (svc *Service) Get(ctx context.Context, req *regionalapipb.GetVpcPeeringRequest) (*regionalapipb.VpcPeering, error) {
	id := req.GetVpcPeeringId()
	vpcpeerings := v1alpha1.VPCPeeringList{}
	err := svc.kubeClient.List(ctx, &vpcpeerings, client.MatchingLabels{
		v1alpha1.LabelKeyBizID: id,
	})
	if client.IgnoreNotFound(err) != nil {
		log.Error("list vpcpeering cr failed", zap.String("id", id), zap.Error(err))
		return nil, bizerr.New(bizerr.K8SClientError, err.Error())
	}
	if len(vpcpeerings.Items) == 0 {
		return nil, bizerr.New(bizerr.ResourceNotFound, "vpcpeering cr not found")
	}
	vpcpeering := &(vpcpeerings.Items[0])
	if !vpcpeering.DeletionTimestamp.IsZero() {
		return nil, bizerr.New(bizerr.ResourceNotFound, "vpcpeering cr is under deleting")
	}
	vpcpeeringpb := new(regionalapipb.VpcPeering)
	if req.GetShowStatusOnly() {
		if err := fillVPCPeeringPBStatus(vpcpeeringpb, vpcpeering); err != nil {
			log.Error("fill vpcpeering pb status failed", zap.Error(err), zap.String("vpcpeering", client.ObjectKeyFromObject(vpcpeering).String()))
			return nil, bizerr.New(bizerr.ConvertError, err.Error())
		}
		return vpcpeeringpb, nil
	}
	if err := fillVPCPeeringPB(vpcpeeringpb, vpcpeering); err != nil {
		log.Error("fill vpcpeering pb failed", zap.Error(err), zap.String("vpcpeering", client.ObjectKeyFromObject(vpcpeering).String()))
		return nil, bizerr.New(bizerr.ConvertError, err.Error())
	}
	return vpcpeeringpb, nil
}

// List
// @Description: list vpcpeerings
// @Parameter ctx
// @Parameter req message.CreateVPCPeeringResp
// @return message.CreateVPCPeeringResp
// @return errors.Error
func (svc *Service) List(ctx context.Context, req *regionalapipb.ListVpcPeeringsRequest) (*regionalapipb.ListVpcPeeringsResponse, error) {
	vpcpeerings := new(v1alpha1.VPCPeeringList)
	// build the lable selector from req
	labelSelector, err := buildLabelSelectorForListVPCPeering(req)
	if err != nil {
		log.Error("build label selector failed", zap.Error(err))
		return nil, bizerr.New(bizerr.ParameterInvalid, err.Error())
	}
	err = svc.kubeClient.List(ctx, vpcpeerings, client.MatchingLabelsSelector{Selector: labelSelector})
	if client.IgnoreNotFound(err) != nil {
		log.Error("list vpcpeering cr failed", zap.Error(err), zap.String("tenantId", req.GetTenantId()), zap.String("projectId", req.GetProjectId()))
		return nil, bizerr.New(bizerr.K8SClientError, err.Error())
	}
	resp := &regionalapipb.ListVpcPeeringsResponse{
		VpcPeerings: make([]*regionalapipb.VpcPeering, 0, len(vpcpeerings.Items)),
	}
	for idx := range vpcpeerings.Items {
		vpcpeering := &(vpcpeerings.Items[idx])
		if !vpcpeering.DeletionTimestamp.IsZero() {
			continue
		}
		// vpc := new(v1alpha1.VPC)
		// vpc.Name = vpcpeering.Spec.SourceVPCRef.Name
		// vpc.Namespace = vpcpeering.Spec.SourceVPCRef.Namespace
		// key := client.ObjectKeyFromObject(vpc)
		// if err := svc.kubeClient.Get(ctx, key, vpc); err != nil {
		// 	svc.logger.Error(ctx, "get vpc failed", "key", key.String(), "err", err)
		// 	return nil, bizerr.New(bizerr.K8SClientError, err.Error())
		// }
		pbdata := &regionalapipb.VpcPeering{}
		if req.GetShowStatusOnly() {
			if err := fillVPCPeeringPBStatus(pbdata, vpcpeering); err != nil {
				log.Error("fill vpcpeering pb status failed", zap.Error(err), zap.String("vpcpeering", client.ObjectKeyFromObject(vpcpeering).String()))
				return nil, bizerr.New(bizerr.K8SClientError, err.Error())
			}
		} else {
			if err := fillVPCPeeringPB(pbdata, vpcpeering); err != nil {
				log.Error("fill vpcpeering pb failed", zap.Error(err), zap.String("vpcpeering", client.ObjectKeyFromObject(vpcpeering).String()))
				return nil, bizerr.New(bizerr.K8SClientError, err.Error())
			}
		}
		resp.VpcPeerings = append(resp.VpcPeerings, pbdata)
	}
	return resp, nil
}

// Accept used to accept a vpcpeering
func (svc *Service) Accept(ctx context.Context, req *regionalapipb.AcceptVpcPeeringRequest) (*emptypb.Empty, error) {
	accepterVPCID := req.AccepterVpcId
	peeringID := req.VpcPeeringId
	requesterCIDR := req.RequesterVpcCidr
	vpcList := &v1alpha1.VPCList{}
	if err := svc.kubeClient.List(ctx, vpcList, client.MatchingLabels{
		v1alpha1.VPCIDKey: accepterVPCID,
	}); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "vpc cr not found")
		}
		log.Error("list vpc cr failed", zap.String("vpcID", accepterVPCID), zap.Error(err))
		return nil, bizerr.New(bizerr.K8SClientError, err.Error())
	}
	if len(vpcList.Items) == 0 {
		return nil, bizerr.New(bizerr.ResourceNotFound, "vpc cr not found")
	}
	vpc := vpcList.Items[0]
	tenantID := vpc.Labels[v1alpha1.TenantIDKey]
	projectID := vpc.Labels[v1alpha1.ProjectIDKey]
	peering := &v1alpha1.VPCPeering{
		ObjectMeta: metav1.ObjectMeta{
			Name:      peeringID,
			Namespace: vpc.Namespace,
			Labels: map[string]string{
				v1alpha1.LabelKeyBizID: peeringID,
				v1alpha1.VPCIDKey:      accepterVPCID,
				v1alpha1.TenantIDKey:   tenantID,
				v1alpha1.ProjectIDKey:  projectID,
				v1alpha1.TagTenantKey:  tenantID,
				v1alpha1.TagProjectKey: projectID,
			},
		},
		Spec: v1alpha1.VPCPeeringSpec{
			Provider: svc.provider,
			Region:   svc.region,
			SourceVPCRef: &v1alpha1.Reference{
				Name:      vpc.Name,
				Namespace: vpc.Namespace,
			},
			AccountRef:             vpc.Spec.AccountRef.DeepCopy(),
			PeeringRole:            v1alpha1.PeeringRoleAccepter,
			VPCPeeringConnectionID: peeringID,
		},
	}
	peering.ObjectMeta.Labels = util.MergeMaps(peering.ObjectMeta.Labels, svc.config.ResourceLabels)
	switch svc.provider {
	case string(constants.AWS):
		peering.Spec.Target.AWS = &v1alpha1.AWS{
			PeerCIDR: requesterCIDR,
		}
	default:
		return nil, bizerr.New(bizerr.ConfigInvalid, "provider not support")
	}
	if !controllerutil.ContainsFinalizer(peering, v1alpha1.FinalizerVPCPeering) {
		controllerutil.AddFinalizer(peering, v1alpha1.FinalizerVPCPeering)
	}
	if err := svc.kubeClient.Create(ctx, peering); err != nil {
		log.Error("create vpcpeering failed", zap.Error(err), zap.String("namespace", peering.Namespace), zap.String("name", peering.Name))
		if !apierrors.IsAlreadyExists(err) {
			return nil, bizerr.New(bizerr.InternalServerError, err.Error())
		}
	}
	return &emptypb.Empty{}, nil
}

func (svc *Service) fillVPCPeeringCR(peering *v1alpha1.VPCPeering, pbdata *regionalapipb.VpcPeering, vpc *v1alpha1.VPC) error {
	// we assume the id given by global service is safe to use as CR name.
	peering.Name = pbdata.GetId()
	peering.Namespace = vpc.Namespace
	peering.Spec.SourceVPCRef = &v1alpha1.Reference{
		Name:      vpc.Name,
		Namespace: vpc.Namespace,
	}
	// TODO(shenjun): how to use the vpc.AccountRef value?
	if vpc.Spec.AccountRef == nil {
		return errors.New("accountRef is nil")
	}
	if vpc.Spec.AccountRef.AccountID == "" {
		return errors.New("accountRef.accountId is empty")
	}
	if vpc.Spec.AccountRef.ProviderConfigName == "" {
		return errors.New("accountRef.providerConfigName is empty")
	}

	// for now we use the vpc's providerconfig
	peering.Spec.AccountRef = vpc.Spec.AccountRef.DeepCopy()
	peering.Spec.Provider = svc.provider
	peering.Spec.Region = svc.region
	sourceAccountID := ""
	switch svc.provider {
	case string(constants.AWS):
		if awsInfo := pbdata.GetAws(); awsInfo == nil {
			return errors.New("aws info is nil")
		} else {
			peerRegion := vpc.Spec.Region
			if awsInfo.GetDestinationRegion() != "" {
				peerRegion = awsInfo.GetDestinationRegion()
			}
			peering.Spec.Target.AWS = &v1alpha1.AWS{
				AccountID:  awsInfo.GetDestinationAccountId(),
				VpcID:      awsInfo.GetDestinationVpcId(),
				PeerRegion: peerRegion,
				PeerCIDR:   pbdata.GetDestinationCidrBlock(),
			}
			sourceAccountID = awsInfo.GetSourceAccountId()
		}
	case string(constants.GCP):
		if gcpInfo := pbdata.GetGcp(); gcpInfo == nil {
			return errors.New("gcp info is nil")
		} else {
			peering.Spec.Target.GCP = &v1alpha1.GCP{
				ProjectID:   gcpInfo.GetDestinationProjectId(),
				PeerNetwork: gcpInfo.GetDestinationNetworkName(),
				PeerCIDR:    pbdata.GetDestinationCidrBlock(),
			}
			sourceAccountID = gcpInfo.GetSourceProjectId()
		}
	}
	// set labels and annotation
	if peering.GetLabels() == nil {
		peering.Labels = map[string]string{}
	}
	peering.Labels[v1alpha1.ProjectIDKey] = pbdata.GetProjectId()
	peering.Labels[v1alpha1.TenantIDKey] = pbdata.GetTenantId()
	peering.Labels[v1alpha1.LabelKeyBizID] = pbdata.GetId()
	// tags for billing
	peering.Labels[v1alpha1.TagTenantKey] = pbdata.GetTenantId()
	peering.Labels[v1alpha1.TagProjectKey] = pbdata.GetProjectId()
	peering.ObjectMeta.Labels = util.MergeMaps(peering.ObjectMeta.Labels, svc.config.ResourceLabels)

	if peering.GetAnnotations() == nil {
		peering.Annotations = map[string]string{}
	}
	peering.Annotations[v1alpha1.AnnotationKeySourceVPCID] = vpc.Status.VPC.VPCID
	peering.Annotations[v1alpha1.AnnotationKeySourceAccountID] = sourceAccountID
	peering.Annotations[v1alpha1.AnnotationKeySourceVPCCIDR] = vpc.Spec.CIDR

	if !controllerutil.ContainsFinalizer(peering, v1alpha1.FinalizerVPCPeering) {
		controllerutil.AddFinalizer(peering, v1alpha1.FinalizerVPCPeering)
	}
	return nil
}

func fillVPCPeeringPBStatus(pbdata *regionalapipb.VpcPeering, vpcpeering *v1alpha1.VPCPeering) error {
	// we cannot use vpcpeering name directly because of migrated vpcpeering has legacy name
	if pbdata.Id == "" {
		pbdata.Id = vpcpeering.Labels[v1alpha1.LabelKeyBizID]
	}
	// FIXME(shenjun): cannot map deleting phase to status
	switch vpcpeering.Status.Phase {
	case v1alpha1.VPCPeeringPhaseActive:
		pbdata.Status = regionalapipb.VpcPeering_ACTIVE
	case v1alpha1.VPCPeeringPhaseFailed:
		pbdata.Status = regionalapipb.VpcPeering_FAILED
	case "", v1alpha1.VPCPeeringPhaseCreating, v1alpha1.VPCPeeringPhasePending:
		pbdata.Status = regionalapipb.VpcPeering_PENDING
	default:
		pbdata.Status = regionalapipb.VpcPeering_STATE_UNSPECIFIED
	}
	return nil
}

func fillVPCPeeringPB(pbdata *regionalapipb.VpcPeering, vpcpeering *v1alpha1.VPCPeering) error {
	// we cannot use vpcpeering name directly because of migrated vpcpeering has legacy name
	if pbdata.Id == "" {
		pbdata.Id = vpcpeering.Labels[v1alpha1.LabelKeyBizID]
	}
	if vpcpeering.GetLabels() != nil {
		pbdata.ProjectId = vpcpeering.GetLabels()[v1alpha1.ProjectIDKey]
		pbdata.TenantId = vpcpeering.GetLabels()[v1alpha1.TenantIDKey]
	}
	var sourceAccountID, sourceVPCID string
	if vpcpeering.GetAnnotations() != nil {
		sourceAccountID = vpcpeering.GetAnnotations()[v1alpha1.AnnotationKeySourceAccountID]
		sourceVPCID = vpcpeering.GetAnnotations()[v1alpha1.AnnotationKeySourceVPCID]
		pbdata.SourceCidrBlock = vpcpeering.GetAnnotations()[v1alpha1.AnnotationKeySourceVPCCIDR]
	}

	switch vpcpeering.Spec.Provider {
	case string(constants.AWS), consts.ProviderTypeEKS:
		awsInfo := new(regionalapipb.VpcPeering_Aws)
		if vpcpeering.Spec.Target.AWS != nil {
			awsInfo.DestinationAccountId = vpcpeering.Spec.Target.AWS.AccountID
			awsInfo.DestinationVpcId = vpcpeering.Spec.Target.AWS.VpcID
			pbdata.DestinationCidrBlock = vpcpeering.Spec.Target.AWS.PeerCIDR
		}
		awsInfo.SourceAccountId = sourceAccountID
		awsInfo.SourceVpcId = sourceVPCID
		awsInfo.VpcPeeringConnectionId = vpcpeering.Status.ExternalName

		pbdata.Info = &regionalapipb.VpcPeering_Aws_{
			Aws: awsInfo,
		}
	case string(constants.GCP), consts.ProviderTypeGKE:
		gcpInfo := new(regionalapipb.VpcPeering_Gcp)
		if vpcpeering.Spec.Target.GCP != nil {
			gcpInfo.DestinationNetworkName = vpcpeering.Spec.Target.GCP.PeerNetwork
			gcpInfo.DestinationProjectId = vpcpeering.Spec.Target.GCP.ProjectID
			pbdata.DestinationCidrBlock = vpcpeering.Spec.Target.GCP.PeerCIDR
		}
		gcpInfo.SourceProjectId = sourceAccountID
		gcpInfo.SourceNetworkName = sourceVPCID
		pbdata.Info = &regionalapipb.VpcPeering_Gcp_{
			Gcp: gcpInfo,
		}
	}
	return fillVPCPeeringPBStatus(pbdata, vpcpeering)
}

func buildLabelSelectorForListVPCPeering(req *regionalapipb.ListVpcPeeringsRequest) (labels.Selector, error) {
	selector := labels.NewSelector()
	if tenantID := req.GetTenantId(); len(tenantID) != 0 {
		r, err := labels.NewRequirement(v1alpha1.TenantIDKey, selection.Equals, []string{tenantID})
		if err != nil {
			return nil, err
		}
		selector = selector.Add(*r)
	}
	if projectID := req.GetProjectId(); len(projectID) != 0 {
		r, err := labels.NewRequirement(v1alpha1.ProjectIDKey, selection.Equals, []string{projectID})
		if err != nil {
			return nil, err
		}
		selector = selector.Add(*r)
	}
	return selector, nil
}