package vpcpeering

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"strings"

	"google.golang.org/protobuf/types/known/emptypb"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"

	dmpbv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	rsconfig "github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/repository"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
)

type Service struct {
	dmpbv1.UnimplementedVpcPeeringServiceServer

	vpcpeeringRW    repository.VPCPeeringReaderWriter
	dataPlaneRW     repository.DataPlaneReaderWriter
	vpcRW           repository.VPCReaderWriter
	externalQueryer ExternalPeeringQueryer
	config          rsconfig.RootConfig
}

func NewService(
	vpcpeeringRW repository.VPCPeeringReaderWriter,
	dataPlaneRW repository.DataPlaneReaderWriter,
	vpcRW repository.VPCReaderWriter,
	externalQueryer ExternalPeeringQueryer,
	config rsconfig.RootConfig,
) dmpbv1.VpcPeeringServiceServer {
	return &Service{
		vpcpeeringRW:    vpcpeeringRW,
		dataPlaneRW:     dataPlaneRW,
		vpcRW:           vpcRW,
		externalQueryer: externalQueryer,
		config:          config,
	}
}

func (s *Service) CreateVpcPeering(ctx context.Context, req *dmpbv1.CreateVpcPeeringRequest) (*dmpbv1.VpcPeering, error) {
	// Validate DataPlane exists
	dp := &v1alpha1.DataPlane{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: s.config.Namespace,
			Name:      req.DataPlaneId,
		},
	}
	if err := s.dataPlaneRW.Get(ctx, client.ObjectKeyFromObject(dp), dp); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "dataplane not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get dataplane error", err)
	}

	sourceVPC := &v1alpha1.VPC{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: dp.Status.Vpc.Namespace,
			Name:      dp.Status.Vpc.Name,
		},
	}
	if err := s.vpcRW.Get(ctx, client.ObjectKeyFromObject(sourceVPC), sourceVPC); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "source vpc not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get source vpc error", err)
	}

	// Get provider from config (as DataPlane doesn't have Provider field)
	provider := s.config.Provider

	// Build VPCPeering CR based on provider
	peeringCR := &v1alpha1.VPCPeering{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: s.config.Namespace,
			// generate name based on dataplane ID and target VPC ID.
			Labels: util.MergeMaps(
				dp.Labels,
				s.config.ResourceLabels,
				map[string]string{
					v1alpha1.DataPlaneIDKey: req.DataPlaneId,
					v1alpha1.VPCNameKey:     sourceVPC.Name,
				},
			),
			// Store source VPC information in annotations for later use in deletion
			Annotations: map[string]string{
				v1alpha1.AnnotationKeySourceVPCID:     sourceVPC.Status.VPC.VPCID,
				v1alpha1.AnnotationKeySourceVPCCIDR:   sourceVPC.Spec.CIDR,
				v1alpha1.AnnotationKeySourceAccountID: dp.Spec.AccountRef.AccountID,
			},
		},
		Spec: v1alpha1.VPCPeeringSpec{
			DataPlane: v1alpha1.Reference{
				Name:      dp.Name,
				Namespace: dp.Namespace,
			},
			Provider: provider,
			Region:   s.config.RegionConfig.Name,
		},
	}

	// Set provider-specific fields
	switch provider {
	case consts.ProviderTypeEKS:
		awsInfo := req.VpcPeering.GetAws()
		if awsInfo == nil {
			return nil, bizerr.New(bizerr.ParameterInvalid, "aws info is required for EKS provider")
		}
		peeringCR.Name = getVPCPeeringName(req.DataPlaneId, awsInfo.DestinationVpcId)
		peeringCR.Spec.Target = v1alpha1.Target{
			AWS: &v1alpha1.AWS{
				AccountID: awsInfo.DestinationAccountId,
				VpcID:     awsInfo.DestinationVpcId,
				PeerCIDR:  req.VpcPeering.DestinationCidrBlock,
				Region:    awsInfo.DestinationRegion,
			},
		}
	case consts.ProviderTypeACK:
		aliInfo := req.VpcPeering.GetAli()
		if aliInfo == nil {
			return nil, bizerr.New(bizerr.ParameterInvalid, "ali info is required for ACK provider")
		}
		peeringCR.Name = getVPCPeeringName(req.DataPlaneId, aliInfo.DestinationVpcId)
		peeringCR.Spec.Target = v1alpha1.Target{
			AliCloud: &v1alpha1.AliCloud{
				UID:      aliInfo.DestinationAccountId,
				VpcID:    aliInfo.DestinationVpcId,
				PeerCIDR: req.VpcPeering.DestinationCidrBlock,
				Region:   aliInfo.DestinationRegion,
			},
		}

	default:
		return nil, bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("unsupported provider: %s", provider))
	}

	// Create VPCPeering CR
	if err := s.vpcpeeringRW.Create(ctx, peeringCR); err != nil {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "create vpc peering error", err)
	}

	// Return initial response
	return convertVPCPeeringToPB(peeringCR), nil
}

func (s *Service) GetVpcPeering(ctx context.Context, req *dmpbv1.GetVpcPeeringRequest) (*dmpbv1.VpcPeering, error) {
	peeringCR := &v1alpha1.VPCPeering{}
	err := s.vpcpeeringRW.Get(ctx, client.ObjectKey{
		Namespace: s.config.Namespace,
		Name:      req.VpcPeeringId,
	}, peeringCR)
	if err == nil {
		return convertVPCPeeringToPB(peeringCR), nil
	}
	if !apierrors.IsNotFound(err) {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get vpc peering error", err)
	}
	if !s.config.BYOC.BYOCPrivateLink.IsEnabled() || s.externalQueryer == nil {
		return nil, bizerr.New(bizerr.ResourceNotFound, "vpc peering not found")
	}

	_, _, vpcCtx, err := s.loadVPCContext(ctx, req.DataPlaneId)
	if err != nil {
		return nil, err
	}

	peering, err := s.externalQueryer.GetByConnectionID(ctx, vpcCtx, req.VpcPeeringId)
	if err != nil {
		return nil, bizerr.Wrap(bizerr.InternalServerError, "query external vpc peering error", err)
	}
	if peering == nil || peering.CurrentSide.VpcID != vpcCtx.VPCID {
		return nil, bizerr.New(bizerr.ResourceNotFound, "vpc peering not found")
	}

	if peering.Provider == "" {
		peering.Provider = vpcCtx.Provider
	}
	return convertExternalPeeringToPB(*peering), nil
}

func (s *Service) DeleteVpcPeering(ctx context.Context, req *dmpbv1.DeleteVpcPeeringRequest) (*emptypb.Empty, error) {
	// Get VPCPeering CR
	peeringCR := &v1alpha1.VPCPeering{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: s.config.Namespace,
			Name:      req.VpcPeeringId,
		},
	}

	if err := s.vpcpeeringRW.Get(ctx, client.ObjectKeyFromObject(peeringCR), peeringCR); err != nil {
		if apierrors.IsNotFound(err) {
			// Already deleted, return success
			return &emptypb.Empty{}, nil
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get vpc peering error", err)
	}

	// Delete VPCPeering CR
	if err := s.vpcpeeringRW.Delete(ctx, peeringCR); err != nil {
		if apierrors.IsNotFound(err) {
			return &emptypb.Empty{}, nil
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "delete vpc peering error", err)
	}

	return &emptypb.Empty{}, nil
}

func (s *Service) ListVpcPeerings(ctx context.Context, req *dmpbv1.ListVpcPeeringsRequest) (*dmpbv1.ListVpcPeeringsResponse, error) {
	peeringList := &v1alpha1.VPCPeeringList{}
	if err := s.vpcpeeringRW.List(ctx, peeringList, client.MatchingLabels{
		v1alpha1.DataPlaneIDKey: req.DataPlaneId,
	}); err != nil {
		return nil, bizerr.Wrap(bizerr.K8SClientError, "list vpc peerings error", err)
	}

	if !s.config.BYOC.BYOCPrivateLink.IsEnabled() || s.externalQueryer == nil {
		pbPeerings := make([]*dmpbv1.VpcPeering, 0, len(peeringList.Items))
		for i := range peeringList.Items {
			pbPeerings = append(pbPeerings, convertVPCPeeringToPB(&peeringList.Items[i]))
		}
		return &dmpbv1.ListVpcPeeringsResponse{VpcPeerings: pbPeerings}, nil
	}

	_, _, vpcCtx, err := s.loadVPCContext(ctx, req.DataPlaneId)
	if err != nil {
		return nil, err
	}

	seenConnectionIDs := make(map[string]struct{}, len(peeringList.Items))
	seenDedupKeys := make(map[string]struct{}, len(peeringList.Items))
	resultItems := make([]ExternalPeering, 0, len(peeringList.Items))
	crByKey := make(map[string]*v1alpha1.VPCPeering, len(peeringList.Items))

	for i := range peeringList.Items {
		cr := &peeringList.Items[i]
		external := externalPeeringFromCR(cr)
		if external.Provider == "" {
			external.Provider = vpcCtx.Provider
		}
		resultItems = append(resultItems, external)
		crByKey[externalPeeringDedupKey(external)] = cr
		if external.ConnectionID != "" {
			seenConnectionIDs[external.ConnectionID] = struct{}{}
		}
		seenDedupKeys[externalPeeringDedupKey(external)] = struct{}{}
	}

	externalPeerings, err := s.externalQueryer.ListByVPC(ctx, vpcCtx)
	if err != nil {
		return nil, err
	}
	for _, peering := range externalPeerings {
		if peering.Provider == "" {
			peering.Provider = vpcCtx.Provider
		}
		if _, ok := seenConnectionIDs[peering.ConnectionID]; ok {
			continue
		}
		key := externalPeeringDedupKey(peering)
		if _, ok := seenDedupKeys[key]; ok {
			continue
		}
		seenDedupKeys[key] = struct{}{}
		if peering.ConnectionID != "" {
			seenConnectionIDs[peering.ConnectionID] = struct{}{}
		}
		resultItems = append(resultItems, peering)
	}

	pbPeerings := make([]*dmpbv1.VpcPeering, 0, len(resultItems))
	for _, peering := range resultItems {
		if s.shouldFilterListPeering(peering) {
			continue
		}

		if cr := crByKey[externalPeeringDedupKey(peering)]; cr != nil {
			pbPeerings = append(pbPeerings, convertVPCPeeringToPB(cr))
			continue
		}
		pbPeerings = append(pbPeerings, convertExternalPeeringToPB(peering))
	}

	return &dmpbv1.ListVpcPeeringsResponse{VpcPeerings: pbPeerings}, nil
}

func (s *Service) loadVPCContext(ctx context.Context, dataplaneID string) (*v1alpha1.DataPlane, *v1alpha1.VPC, VPCContext, error) {
	dp := &v1alpha1.DataPlane{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: s.config.Namespace,
			Name:      dataplaneID,
		},
	}
	if err := s.dataPlaneRW.Get(ctx, client.ObjectKeyFromObject(dp), dp); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, nil, VPCContext{}, bizerr.New(bizerr.ResourceNotFound, "dataplane not found")
		}
		return nil, nil, VPCContext{}, bizerr.Wrap(bizerr.K8SClientError, "get dataplane error", err)
	}
	if dp.Status.Vpc == nil {
		return nil, nil, VPCContext{}, bizerr.New(bizerr.ResourceNotFound, "dataplane vpc not found")
	}

	vpc := &v1alpha1.VPC{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: dp.Status.Vpc.Namespace,
			Name:      dp.Status.Vpc.Name,
		},
	}
	if err := s.vpcRW.Get(ctx, client.ObjectKeyFromObject(vpc), vpc); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, nil, VPCContext{}, bizerr.New(bizerr.ResourceNotFound, "source vpc not found")
		}
		return nil, nil, VPCContext{}, bizerr.Wrap(bizerr.K8SClientError, "get source vpc error", err)
	}

	return dp, vpc, VPCContext{
		Provider:    s.config.Provider,
		Region:      vpc.Spec.Region,
		VPCID:       vpc.Status.VPC.VPCID,
		VPCCIDR:     vpc.Spec.CIDR,
		AccountID:   dp.Spec.AccountRef.AccountID,
		AccountRole: dp.Spec.AccountRef.Role,
	}, nil
}

func (s *Service) shouldFilterListPeering(peering ExternalPeering) bool {
	o11yVPCID := strings.TrimSpace(s.config.O11Y.VPCID)
	if o11yVPCID == "" {
		return false
	}
	return peering.CurrentSide.VpcID == o11yVPCID || peering.DestinationSide.VpcID == o11yVPCID
}

// getVPCPeeringName generates VPCPeering CR name based on dataplane ID and target VPC ID,
// because a dataplane can have multiple VPC peerings with different target VPCs.
// Uses SHA256 hash of target VPC ID to ensure unique and valid Kubernetes resource names.
func getVPCPeeringName(dataplaneID, targetVPC string) string {
	// Hash the target VPC ID and take first 8 characters for uniqueness and safety
	hash := sha256.Sum256([]byte(targetVPC))
	shortHash := hex.EncodeToString(hash[:4]) // 8 characters
	return fmt.Sprintf("%s-%s", dataplaneID, shortHash)
}

// convertVPCPeeringToPB converts VPCPeering CR to protobuf message
func convertVPCPeeringToPB(cr *v1alpha1.VPCPeering) *dmpbv1.VpcPeering {
	pb := &dmpbv1.VpcPeering{
		Id:      cr.Name,
		Status:  convertPhaseToPBStatus(cr.Status.Phase),
		Message: cr.Status.Reason,
	}

	// Set source CIDR if available
	if cidr := cr.Annotations[v1alpha1.AnnotationKeySourceVPCCIDR]; cidr != "" {
		pb.SourceCidrBlock = cidr
	}

	// Set provider-specific info
	if cr.Spec.Target.AWS != nil {
		pb.DestinationCidrBlock = cr.Spec.Target.AWS.PeerCIDR
		awsInfo := &dmpbv1.VpcPeering_Aws{
			DestinationAccountId:   cr.Spec.Target.AWS.AccountID,
			DestinationVpcId:       cr.Spec.Target.AWS.VpcID,
			DestinationRegion:      cr.Spec.Target.AWS.Region,
			SourceAccountId:        cr.Annotations[v1alpha1.AnnotationKeySourceAccountID],
			SourceRegion:           cr.Spec.Region,
			SourceVpcId:            cr.Annotations[v1alpha1.AnnotationKeySourceVPCID],
			VpcPeeringConnectionId: cr.Status.ExternalName,
		}
		pb.Info = &dmpbv1.VpcPeering_Aws_{Aws: awsInfo}
	} else if cr.Spec.Target.AliCloud != nil {
		pb.DestinationCidrBlock = cr.Spec.Target.AliCloud.PeerCIDR
		aliInfo := &dmpbv1.VpcPeering_Alicloud{
			DestinationAccountId:   cr.Spec.Target.AliCloud.UID,
			DestinationVpcId:       cr.Spec.Target.AliCloud.VpcID,
			DestinationRegion:      cr.Spec.Target.AliCloud.Region,
			SourceAccountId:        cr.Annotations[v1alpha1.AnnotationKeySourceAccountID],
			SourceRegion:           cr.Spec.Region,
			SourceVpcId:            cr.Annotations[v1alpha1.AnnotationKeySourceVPCID],
			VpcPeeringConnectionId: cr.Status.ExternalName,
		}
		pb.Info = &dmpbv1.VpcPeering_Ali{Ali: aliInfo}
	}

	return pb
}

func convertExternalPeeringToPB(peering ExternalPeering) *dmpbv1.VpcPeering {
	pb := &dmpbv1.VpcPeering{
		Id:                   peering.ConnectionID,
		SourceCidrBlock:      peering.CurrentSide.CIDR,
		DestinationCidrBlock: peering.DestinationSide.CIDR,
		Message:              peering.Message,
		Status:               convertExternalStatusToPB(peering.Status),
	}

	switch peering.Provider {
	case consts.ProviderTypeACK:
		pb.Info = &dmpbv1.VpcPeering_Ali{Ali: &dmpbv1.VpcPeering_Alicloud{
			SourceAccountId:        peering.CurrentSide.AccountID,
			SourceVpcId:            peering.CurrentSide.VpcID,
			SourceRegion:           peering.CurrentSide.Region,
			DestinationAccountId:   peering.DestinationSide.AccountID,
			DestinationVpcId:       peering.DestinationSide.VpcID,
			DestinationRegion:      peering.DestinationSide.Region,
			VpcPeeringConnectionId: peering.ConnectionID,
		}}
	default:
		pb.Info = &dmpbv1.VpcPeering_Aws_{Aws: &dmpbv1.VpcPeering_Aws{
			SourceAccountId:        peering.CurrentSide.AccountID,
			SourceVpcId:            peering.CurrentSide.VpcID,
			SourceRegion:           peering.CurrentSide.Region,
			DestinationAccountId:   peering.DestinationSide.AccountID,
			DestinationVpcId:       peering.DestinationSide.VpcID,
			DestinationRegion:      peering.DestinationSide.Region,
			VpcPeeringConnectionId: peering.ConnectionID,
		}}
	}

	return pb
}

// convertPhaseToPBStatus converts VPCPeeringPhase to protobuf Status
func convertPhaseToPBStatus(phase v1alpha1.VPCPeeringPhase) dmpbv1.VpcPeering_Status {
	switch phase {
	case v1alpha1.VPCPeeringPhasePending:
		return dmpbv1.VpcPeering_PENDING
	case v1alpha1.VPCPeeringPhaseActive:
		return dmpbv1.VpcPeering_ACTIVE
	case v1alpha1.VPCPeeringPhaseFailed:
		return dmpbv1.VpcPeering_FAILED
	default:
		return dmpbv1.VpcPeering_STATE_UNSPECIFIED
	}
}

func convertExternalStatusToPB(status string) dmpbv1.VpcPeering_Status {
	switch strings.ToLower(status) {
	case "pending", "pending-acceptance", "initiating-request", "provisioning", "creating", "accepting":
		return dmpbv1.VpcPeering_PENDING
	case "active", "activated", "available", "connected":
		return dmpbv1.VpcPeering_ACTIVE
	case "failed", "rejected":
		return dmpbv1.VpcPeering_FAILED
	default:
		return dmpbv1.VpcPeering_STATE_UNSPECIFIED
	}
}