package privatelink

import (
	"context"
	"net"
	"regexp"
	"strconv"

	"github.com/samber/lo"
	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/privatelink"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/tidbgroup"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/resourcepoolpolicy"
	servicecommon "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/common"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

var (
	validAWSEndpointID      = regexp.MustCompile("^vpce-[0-9a-f]{17}$")
	validAliCloudEndpointID = regexp.MustCompile("^ep-.*[0-9a-f]$")
)

func (s *service) CreatePrivateLinkConnection(
	ctx context.Context,
	req *globalngv1.CreatePrivateLinkConnectionRequest,
) (*entity.PrivateLinkConnection, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "privatelink.CreatePrivateLinkConnection")
	defer span.End()

	tidbGroupID := req.GetTidbGroupId()
	if tidbGroupID == 0 {
		tidbGroups, err := s.tidbGroupRepo.Query(ctx, &tidbgroup.Filter{
			ClusterIds: []uint64{req.GetClusterId()},
			Default:    lo.ToPtr(true),
		})
		if err != nil {
			return nil, err
		}
		if len(tidbGroups) == 0 {
			return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "default tidb group not found")
		}
		tidbGroupID = tidbGroups[0].Id
	} else if _, err := s.tidbGroupRepo.Load(ctx, req.GetClusterId(), tidbGroupID); err != nil {
		return nil, err
	}
	cluster, err := s.clusterRepo.Get(ctx, req.GetClusterId(), false)
	if err != nil {
		return nil, err
	}
	if berr := servicecommon.CheckClusterResourcePoolLifecycle(
		ctx, s.resourcePoolRepo, cluster, resourcepoolpolicy.OperationUpdateNetwork,
	); berr != nil {
		return nil, berr
	}
	cloudProvider := cluster.GetCloudProviderId()
	if err := s.checkParams(req.GetPrivateLinkConnection(), cloudProvider); err != nil {
		return nil, err
	}
	region, err := s.regionRepo.Get(ctx, cluster.RegionId)
	if err != nil {
		return nil, err
	}
	psEntities, err := s.privatelinkRepo.ListPrivateLinkServices(ctx, &privatelink.Filter{
		ClusterID: req.GetClusterId(), TidbGroupID: tidbGroupID,
	})
	if err != nil {
		return nil, err
	}
	if len(psEntities) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "private link service not found")
	}
	psEntity := psEntities[0]
	privateLinkService, err := s.regionalCli.GetPrivateLinkService(
		ctx,
		region,
		&clusterngv1beta.GetPrivateLinkServiceRequest{
			ClusterId:            strconv.FormatUint(cluster.ClusterId, 10),
			TidbGroupId:          strconv.FormatUint(tidbGroupID, 10),
			PrivateLinkServiceId: strconv.FormatUint(psEntity.PrivateLinkServiceID, 10),
		},
	)
	if err != nil {
		return nil, err
	}
	if err := s.check(ctx, req, cloudProvider, cluster, privateLinkService, tidbGroupID); err != nil {
		return nil, err
	}

	var out *entity.PrivateLinkConnection
	terr := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		if berr := servicecommon.LockClusterResourcePoolLifecycle(
			ctx, s.resourcePoolRepo, cluster, resourcepoolpolicy.OperationUpdateNetwork,
		); berr != nil {
			return berr
		}
		ev, err := entity.NewClusterEvent(ctx, cluster, clusterevent.EventTypeCreatingPrivateLink, "CLUSTER_STATUS_MODIFYING")
		if err != nil {
			return err
		}
		if err := s.eventCommitter.Write(ctx, clusterevent.TopicClusterEvent, ev); err != nil {
			return err
		}
		// Create
		privateLinkConnectionID := s.idGeneratorFacade.MustNew()
		privateLinkConnectionEntity := &entity.PrivateLinkConnection{
			PrivateLinkConnection: &globalngv1.PrivateLinkConnection{
				PrivateLinkConnectionId: strconv.FormatUint(privateLinkConnectionID, 10),
				ClusterId:               req.GetClusterId(),
				TidbGroupId:             tidbGroupID,
				RegionId:                region.GetId(),
				RegionDisplayName:       region.GetDisplayName(),
				CloudProviderId:         region.GetCloudProviderId(),
				PrivateLinkService:      psEntity.GetServiceName(),
				PrivateLinkServiceState: coverToGlobalPrivateLinkServiceState(privateLinkService.GetStatus()),
				EndpointType:            req.GetPrivateLinkConnection().GetEndpointType(),
				PrivateIpAddress:        req.GetPrivateLinkConnection().GetPrivateIpAddress(),
				EndpointId:              req.GetPrivateLinkConnection().GetEndpointId(),
				PrivateDomainName:       req.GetPrivateLinkConnection().GetPrivateDomainName(),
			},
		}

		if err := s.privatelinkRepo.CreatePrivateLinkConnection(ctx, privateLinkConnectionEntity); err != nil {
			if err.ErrCode() == bizerr.ErrCodeRecordExist {
				return err.OverwriteMessage(
					"endpoint %s already exists for cluster %d",
					req.GetPrivateLinkConnection().GetEndpointId(),
					req.GetClusterId(),
				)
			}
			return err
		}

		out = privateLinkConnectionEntity
		out.State = globalngv1.PrivateLinkConnection_PENDING

		// Start async reconciliation workflow.
		workFlowErr := s.flow.RunDagWithBizId(
			ctx,
			common.CreatePrivateLinkConnection,
			strconv.FormatUint(cluster.OrganizationId, 10),
			"0",
			strconv.FormatUint(cluster.ClusterId, 10),
			map[string]string{
				workflow.KeyTraceId:               workflow.GetTraceId(ctx),
				common.KeyClusterId:               strconv.FormatUint(cluster.ClusterId, 10),
				common.KeyPrivateLinkConnectionId: strconv.FormatUint(privateLinkConnectionID, 10),
				common.KeyPrivateLinkServiceId:    strconv.FormatUint(psEntity.PrivateLinkServiceID, 10),
			},
		)
		if workFlowErr != nil {
			return bizerr.WrapErrorWithCode(workFlowErr, bizerr.ErrCodeUnknown, "create async workflow failed")
		}

		return nil
	})
	if terr != nil {
		return nil, bizerr.WrapError(terr, "create privatelink endpoint failed")
	}

	return out, nil
}

func (s *service) check(
	ctx context.Context,
	req *globalngv1.CreatePrivateLinkConnectionRequest,
	cloudProvider globalcommonv1.CloudProvider_Name,
	cluster *entity.Cluster,
	privateLinkService *clusterngv1beta.PrivateLinkService,
	tidbGroupId uint64,
) bizerr.BizError {
	if err := s.checkParams(req.GetPrivateLinkConnection(), cloudProvider); err != nil {
		return err
	}

	if err := s.checkPrivateLinkServiceStatus(cluster, privateLinkService); err != nil {
		return err
	}

	resp, err := s.privatelinkRepo.ListPrivateLinkConnections(ctx, &privatelink.Filter{
		ClusterID:   req.GetClusterId(),
		TidbGroupID: tidbGroupId,
	})
	if err != nil {
		return err
	}

	if err := s.checkUniqueEndpointPerCluster(req, resp); err != nil {
		return err
	}

	if err := s.checkMaxConnectionPerGroup(cloudProvider, resp, tidbGroupId, 1); err != nil {
		return err
	}

	return nil
}

func (s *service) checkParams(
	conn *globalngv1.PrivateLinkConnection,
	cloudProvider globalcommonv1.CloudProvider_Name,
) bizerr.BizError {
	switch cloudProvider {
	case globalcommonv1.CloudProvider_AWS:
		if !validAWSEndpointID.MatchString(conn.GetEndpointId()) {
			return bizerr.New(bizerr.ErrCodeParams, "invalid endpoint id")
		}
	case globalcommonv1.CloudProvider_AZURE:
		if conn.GetEndpointId() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "invalid endpoint id")
		}
		if net.ParseIP(conn.GetPrivateIpAddress()) == nil {
			return bizerr.New(bizerr.ErrCodeParams, "invalid private ip address")
		}
	case globalcommonv1.CloudProvider_ALICLOUD:
		if !validAliCloudEndpointID.MatchString(conn.GetEndpointId()) {
			return bizerr.New(bizerr.ErrCodeParams, "invalid endpoint id")
		}
	default:
		return bizerr.New(bizerr.ErrCodeParams, "invalid provider: %s", cloudProvider)
	}

	if conn.GetEndpointType() == globalngv1.PrivateLinkConnection_ENDPOINT_TYPE_UNSPECIFIED {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"unspecified endpoint type: %s",
			conn.GetEndpointType(),
		)
	}

	return nil
}

func (s *service) checkPrivateLinkServiceStatus(
	cluster *entity.Cluster,
	privateLinkService *clusterngv1beta.PrivateLinkService,
) bizerr.BizError {
	// Cluster state must be AVAILABLE or SCALING.
	if cluster.State != globalngv1.Cluster_AVAILABLE {
		return bizerr.New(
			bizerr.ErrCodePrivateLinkServiceNotAllowed,
			"unable to create PrivateLink endpoint when cluster status is %s",
			cluster.State.String(),
		)
	}

	// PrivateLink service status must be ACTIVE.
	if privateLinkService.Status != clusterngv1beta.PrivateLinkService_ACTIVE {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"unable to create endpoint when PrivateLink service status is: %s",
			privateLinkService.Status.String(),
		)
	}

	return nil
}

func (s *service) checkMaxConnectionPerGroup(
	cloudProvider globalcommonv1.CloudProvider_Name,
	resp []*entity.PrivateLinkConnection,
	tidbGroupId uint64,
	addCount int,
) bizerr.BizError {
	var limit int
	switch cloudProvider {
	case globalcommonv1.CloudProvider_AWS:
		limit = s.cfg.BizConfig.AwsPrivateLinkEndpointLimitPerGroup
	case globalcommonv1.CloudProvider_GCP:
		limit = s.cfg.BizConfig.GcpPrivateLinkEndpointLimitPerGroup
	case globalcommonv1.CloudProvider_AZURE:
		limit = s.cfg.BizConfig.AzurePrivateLinkEndpointLimitPerGroup
	case globalcommonv1.CloudProvider_ALICLOUD:
		limit = s.cfg.BizConfig.AliCloudPrivateLinkEndpointLimitPerGroup
	default:
		return bizerr.New(bizerr.ErrCodeParams, "invalid cloud provider: %s", cloudProvider)
	}

	count := lo.CountBy(resp, func(item *entity.PrivateLinkConnection) bool {
		return item.GetTidbGroupId() == tidbGroupId
	})

	if limit != 0 && count+addCount > limit {
		return bizerr.New(bizerr.ErrCodeParams, "too many endpoints, each TiDB group can have up to %d", limit)
	}

	return nil
}

func (s *service) checkUniqueEndpointPerCluster(
	req *globalngv1.CreatePrivateLinkConnectionRequest,
	resp []*entity.PrivateLinkConnection,
) bizerr.BizError {
	_, ok := lo.Find(resp, func(item *entity.PrivateLinkConnection) bool {
		return item.EndpointId == req.GetPrivateLinkConnection().GetEndpointId()
	})
	if ok {
		return bizerr.New(
			bizerr.ErrCodeRecordExist,
			"endpoint %s already exists for cluster %d",
			req.GetPrivateLinkConnection().GetEndpointId(),
			req.GetClusterId(),
		)
	}

	return nil
}