package vpc

import (
	"context"
	"strconv"

	regionaldedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	cidrrepo "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/cidr"
	clusterrepo "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/cluster"
	endpointrepo "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/endpoint"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/enum"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	ErrMsgVPCPeeringConflictWithCurrentPeeringCidr = "The %s VPC CIDR you filled in conflicts with the current Peering VPC CIDR, and you cannot create a VPC Peering"
	ErrMsgVPCPeeringConflictWithRegionCidr         = "The %s VPC CIDR you filled in conflicts with %s(%s), and you cannot create a VPC Peering"
	ErrMsgVPCPeeringCreatePrerequisite             = "You should create at least one cluster in that region before creating a VPC Peering"
)

func (s *serviceImpl) CreateVpcPeering(
	ctx context.Context,
	params *globaldedicatedv1.VpcPeering,
) (*entity.VpcPeering, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Domain.CreateVpcPeeringSvc")
	defer span.End()

	// Check the required fields.
	switch params.CloudProvider {
	case globalcommonv1.CloudProvider_AWS:
		if params.Aws == nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "aws peering type must have AWS params")
		}
		if params.Aws.GetCustomerAccountId() == "" {
			return nil, bizerr.New(bizerr.ErrCodeParams, "aws peering type must have AWS account id")
		}
		if params.Aws.GetCustomerVpcId() == "" {
			return nil, bizerr.New(bizerr.ErrCodeParams, "aws peering type must have AWS vpc id")
		}
	case globalcommonv1.CloudProvider_GCP:
		if params.Gcp == nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "gcp peering type must have GCP params")
		}
		if params.Gcp.GetCustomerProjectId() == "" {
			return nil, bizerr.New(bizerr.ErrCodeParams, "gcp peering type must have GCP project id")
		}
		if params.Gcp.GetCustomerNetworkName() == "" {
			return nil, bizerr.New(bizerr.ErrCodeParams, "gcp peering type must have GCP network name")
		}
	case globalcommonv1.CloudProvider_AZURE:
		// FIXME: @SabaPing: add azure peering params.
	}

	regionName, err := enum.FromString[globalcommonv1.Region_Name](params.GetTidbCloudRegion())
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid region")
	}

	transErr := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		regionEntity, err := s.regionRepo.SelectRegionById(
			ctx,
			globalcommonv1.Region_Name(regionName),
			params.GetOrganizationId(),
		)
		if err != nil {
			return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid region")
		}

		if regionEntity.CloudProviderId != params.GetCloudProvider() {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"unsupported region %s in %s cloud provider",
				params.GetTidbCloudRegion(),
				params.CloudProvider,
			)
		}
		if err := s.ensureVpcPeeringCreatePrerequisite(ctx, params.GetOrganizationId(), params.GetProjectId(), regionEntity); err != nil {
			return err
		}

		listVpcsRequest := &regionaldedicatedv1.ListVpcsRequest{
			TenantId:  strconv.FormatUint(params.GetOrganizationId(), 10),
			ProjectId: strconv.FormatUint(params.GetProjectId(), 10),
			ClusterId: "", // leave it empty
			Statuses: []regionaldedicatedv1.Vpc_Status{
				regionaldedicatedv1.Vpc_ACTIVE,
				regionaldedicatedv1.Vpc_CREATING,
				regionaldedicatedv1.Vpc_FAILED,
			},
		}
		listRes, err := s.regionServerCli.ListVpcs(ctx, regionEntity, listVpcsRequest)
		if err != nil {
			return err
		}
		if len(listRes.Vpcs) == 0 {
			return bizerr.New(bizerr.ErrCodeVPCPeeringCreatePrerequisite, ErrMsgVPCPeeringCreatePrerequisite)
		}
		if len(listRes.Vpcs) != 1 {
			return bizerr.New(bizerr.ErrCodeUnknown, "number of vpcs is not 1")
		}
		vpc := listRes.Vpcs[0]
		secondaryCidr, err := s.cidrRepo.QuerySecondaryCidrs(ctx, &cidrrepo.Filter{
			VpcIDs: []string{vpc.GetVpcId()},
		}, nil)
		if err != nil {
			return err
		}
		if len(secondaryCidr) > 0 {
			return bizerr.New(bizerr.ErrCodeParams, "vpc peering is not supported for vpc that has secondary cidr")
		}

		if err := s.checkPeerCidr(ctx, params, params.GetCloudProvider(), vpc); err != nil {
			return err
		}

		params.Id = s.idGeneratorFacade.MustNew()

		if err := s.vpcPeeringRepo.Create(ctx, params); err != nil {
			return err
		}

		workFlowErr := s.workflow.RunDag(
			ctx,
			common.CreateVpcPeering,
			strconv.FormatUint(params.GetOrganizationId(), 10),
			strconv.FormatUint(params.GetProjectId(), 10),
			map[string]string{
				workflow.KeyTraceId:         workflow.GetTraceId(ctx),
				common.KeyVpcPeeringId:      strconv.FormatUint(params.GetId(), 10),
				common.KeyCloudProviderName: params.CloudProvider.String(),
			},
		)
		if workFlowErr != nil {
			return bizerr.WrapErrorWithCode(workFlowErr, bizerr.ErrCodeUnknown, "create async workflow failed")
		}
		return nil
	})
	if transErr != nil {
		return nil, bizerr.WrapError(transErr, "failed to create vpc peering")
	}

	return params, nil
}

func (s *serviceImpl) ensureVpcPeeringCreatePrerequisite(
	ctx context.Context,
	orgID uint64,
	projectID uint64,
	region *entity.Region,
) bizerr.BizError {
	clusters, err := s.clusterRepo.Query(ctx, &clusterrepo.Filter{
		CloudProviderNames: []entity.CloudProviderName{region.CloudProviderId},
		RegionNames:        []entity.RegionName{region.Id},
		OrganizationIds:    []uint64{orgID},
		ProjectIds:         []uint64{projectID},
		ExcludeStates: []entity.State{
			globaldedicatedv1.Cluster_STATE_UNSPECIFIED,
			globaldedicatedv1.Cluster_DELETED,
		},
	})
	if err != nil {
		return err
	}
	if len(clusters) == 0 {
		return bizerr.New(bizerr.ErrCodeVPCPeeringCreatePrerequisite, ErrMsgVPCPeeringCreatePrerequisite)
	}

	return nil
}

func (s *serviceImpl) checkPeerCidr(
	ctx context.Context,
	params *entity.VpcPeering,
	cloudProviderName globalcommonv1.CloudProvider_Name,
	vpc *regionaldedicatedv1.Vpc,
) bizerr.BizError {
	if params.GetType() == globaldedicatedv1.VpcPeering_CUSTOMER {
		peeringList, err := s.vpcPeeringRepo.Query(ctx, &endpointrepo.VpcPeeringFilter{
			CloudProviderNames: []globalcommonv1.CloudProvider_Name{cloudProviderName},
			ProjectIds:         []uint64{params.GetProjectId()},
			Types: []globaldedicatedv1.VpcPeering_Type{
				globaldedicatedv1.VpcPeering_CUSTOMER,
			}, // make sure customer vpcCidr is unique in any region of the project
		},
			nil,
			nil,
		)
		if err != nil {
			return err
		}
		// check peerVpcCidr with used peerVpcCidr
		for _, l := range peeringList {
			if l.CustomerCidrBlock != "" {
				if err := IsOverlapCidr(
					ctx,
					params.GetCustomerCidrBlock(),
					l.GetCustomerCidrBlock(),
					l.GetCustomerRegion(),
					CidrTypeCurrent,
					params.CloudProvider.String(),
				); err != nil {
					return err
				}
			}
		}
	}
	// check peerVpcCidr with vpc cidr
	if err := IsOverlapCidr(
		ctx,
		params.GetCustomerCidrBlock(),
		vpc.CidrBlock,
		params.GetTidbCloudRegion(),
		CidrTypeRegion,
		params.CloudProvider.String(),
	); err != nil {
		return err
	}

	return nil
}