package privatelink

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

	"github.com/samber/lo"
	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/adapter/regionserver"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	"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/workflow"
)

var (
	validAWSEndpointID   = regexp.MustCompile("^vpce-[0-9a-f]{17}$")
	validGCPEndpointID   = regexp.MustCompile("^projects/.+/regions/.+/forwardingRules/.+$")
	validAzureEndpointID = regexp.MustCompile(
		`^/subscriptions/[^/]+/resourceGroups/[^/]+/providers/Microsoft\.Network/privateEndpoints/[^/]+$`,
	)
	validEndpointID = regexp.MustCompile(`^\d+$`)
)

func (svc *service) CreatePrivateLinkEndpoint(
	ctx context.Context,
	req *globaldedicatedv1.CreatePrivateLinkEndpointRequest,
) (*globaldedicatedv1.PrivateLinkEndpoint, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Domain.CreatePrivateLinkEndpointSvc")
	defer span.End()

	var out *globaldedicatedv1.PrivateLinkEndpoint

	err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		tidbGroupID := req.GetTidbGroupId()
		if tidbGroupID == 0 {
			tidbGroup, err := svc.tidbGroupRepo.LoadDefault(ctx, req.GetClusterId())
			if err != nil {
				return err
			}
			tidbGroupID = tidbGroup.Id
		} else {
			// make sure tidbGroupID is owned by the cluster
			_, err := svc.tidbGroupRepo.Load(ctx, req.GetClusterId(), tidbGroupID)
			if err != nil {
				return err
			}
		}

		cluster, err := svc.clusterRepo.Get(ctx, req.GetClusterId(), false)
		if err != nil {
			return err
		}
		cloudProvider := cluster.CloudProviderId

		region, err := svc.regionRepo.SelectRegionById(ctx, cluster.RegionId, cluster.OrganizationId)
		if err != nil {
			return err
		}

		privateLinkService, err := svc.regionServerCli.GetPrivateLinkService(
			ctx,
			region,
			&regionserver.GetPrivateLinkServiceRequest{
				ClusterId:   strconv.FormatUint(cluster.ClusterId, 10),
				TidbGroupId: strconv.FormatUint(tidbGroupID, 10),
			},
		)
		if err != nil {
			return err
		}

		// Check
		if err := svc.check(ctx, req, cloudProvider, cluster, privateLinkService, tidbGroupID); err != nil {
			return err
		}

		// Create
		privateLinkEndpointID := svc.idGeneratorFacade.MustNew()
		privateLinkEndpointEntity := &entity.PrivateLinkEndpoint{
			ID:               privateLinkEndpointID,
			TenantID:         req.GetOrgId(),
			ProjectID:        req.GetProjectId(),
			ClusterID:        req.GetClusterId(),
			TidbGroupID:      tidbGroupID,
			Type:             req.GetPrivateLinkEndpoint().GetType(),
			EndpointID:       req.PrivateLinkEndpoint.GetEndpointName(),
			PrivateIpAddress: req.GetPrivateLinkEndpoint().GetPrivateIpAddress(),
			CloudProvider:    cloudProvider.String(),
		}
		// Validate only for Azure, because AWS and GCP don't support validate only on region server.
		if req.GetValidateOnly() && cloudProvider == globalcommonv1.CloudProvider_AZURE {
			connection, err := buildPrivateLinkConnection(req, cloudProvider)
			if err != nil {
				return err
			}
			_, err = svc.regionServerCli.UpdatePrivateLinkService(
				ctx,
				region,
				&regionserver.UpdatePrivateLinkServiceRequest{
					ClusterId:   strconv.FormatUint(cluster.ClusterId, 10),
					TidbGroupId: strconv.FormatUint(tidbGroupID, 10),
					PrivateLinkConnections: []*regionaldedicatedv1.PrivateLinkService_Connection{
						connection,
					},
					ValidateOnly: req.GetValidateOnly(),
				},
			)
			return err
		}

		if err := svc.privateLinkEndpointRepo.Create(ctx, privateLinkEndpointEntity); err != nil {
			if err.ErrCode() == bizerr.ErrCodeRecordExist {
				return err.OverwriteMessage(
					"endpoint %s already exists for cluster %d",
					req.GetPrivateLinkEndpoint().GetEndpointName(),
					req.GetClusterId(),
				)
			}
			return err
		}

		out = req.GetPrivateLinkEndpoint()
		out.Id = privateLinkEndpointID
		out.Status = globaldedicatedv1.PrivateLinkEndpoint_PENDING

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

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

	return out, nil
}

func buildPrivateLinkConnection(
	req *globaldedicatedv1.CreatePrivateLinkEndpointRequest,
	cloudProvider globalcommonv1.CloudProvider_Name,
) (*regionaldedicatedv1.PrivateLinkService_Connection, bizerr.BizError) {
	switch cloudProvider {
	case globalcommonv1.CloudProvider_AZURE:
		return &regionaldedicatedv1.PrivateLinkService_Connection{
			Info: &regionaldedicatedv1.PrivateLinkService_Connection_Azure_{
				Azure: &regionaldedicatedv1.PrivateLinkService_Connection_Azure{
					PrivateEndpointResourceId: req.GetPrivateLinkEndpoint().GetEndpointName(),
					PrivateIpAddress:          req.GetPrivateLinkEndpoint().GetPrivateIpAddress(),
					Host:                      req.PrivateLinkEndpoint.GetHost(),
				},
			},
		}, nil
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid cloud provider: %s", cloudProvider)
	}
}

func (svc *service) check(
	ctx context.Context,
	req *globaldedicatedv1.CreatePrivateLinkEndpointRequest,
	cloudProvider globalcommonv1.CloudProvider_Name,
	cluster *entity.Cluster,
	privateLinkService *regionaldedicatedv1.PrivateLinkService,
	tidbGroupId uint64,
) bizerr.BizError {
	if err := svc.checkParams(req, cloudProvider); err != nil {
		return err
	}

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

	resp, err := svc.privateLinkEndpointRepo.Query(ctx, &endpoint.PrivateLinkEndpointFilter{
		ClusterIDs: []uint64{cluster.GetClusterId()},
	})
	if err != nil {
		return err
	}

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

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

	return nil
}

func (svc *service) checkParams(
	req *globaldedicatedv1.CreatePrivateLinkEndpointRequest,
	cloudProvider globalcommonv1.CloudProvider_Name,
) bizerr.BizError {
	switch cloudProvider {
	case globalcommonv1.CloudProvider_AWS:
		if !validAWSEndpointID.MatchString(req.GetPrivateLinkEndpoint().GetEndpointName()) {
			return bizerr.New(bizerr.ErrCodeParams, "invalid endpoint name")
		}
	case globalcommonv1.CloudProvider_GCP:
		if !validGCPEndpointID.MatchString(req.GetPrivateLinkEndpoint().GetEndpointName()) {
			return bizerr.New(bizerr.ErrCodeParams, "invalid endpoint name")
		}
	case globalcommonv1.CloudProvider_AZURE:
		if !validAzureEndpointID.MatchString(req.GetPrivateLinkEndpoint().GetEndpointName()) {
			return bizerr.New(bizerr.ErrCodeParams, "invalid endpoint name")
		}
		if req.GetPrivateLinkEndpoint().GetPrivateIpAddress() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "private IP address is required for Azure")
		}
		ip := net.ParseIP(req.GetPrivateLinkEndpoint().GetPrivateIpAddress())
		if ip == nil {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"invalid private IP address: %s",
				req.GetPrivateLinkEndpoint().GetPrivateIpAddress(),
			)
		}
	default:
		return bizerr.New(bizerr.ErrCodeParams, "invalid provider: %s", cloudProvider)
	}

	if req.GetPrivateLinkEndpoint().GetType() == globaldedicatedv1.PrivateLinkEndpoint_TYPE_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "unspecified endpoint type: %s", req.GetPrivateLinkEndpoint().GetType())
	}

	return nil
}

func (svc *service) checkPrivateLinkServiceStatus(
	cluster *entity.Cluster,
	privateLinkService *regionaldedicatedv1.PrivateLinkService,
) bizerr.BizError {
	// Cluster state must be AVAILABLE or SCALING.
	if cluster.State != globaldedicatedv1.Cluster_AVAILABLE && cluster.State != globaldedicatedv1.Cluster_SCALING {
		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 != regionaldedicatedv1.PrivateLinkService_ACTIVE {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"unable to create endpoint when PrivateLink service status is: %s",
			privateLinkService.Status.String(),
		)
	}

	return nil
}

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

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

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

	return nil
}

func (svc *service) checkUniqueEndpointPerCluster(
	req *globaldedicatedv1.CreatePrivateLinkEndpointRequest,
	resp []*entity.PrivateLinkEndpoint,
) bizerr.BizError {
	_, ok := lo.Find(resp, func(item *entity.PrivateLinkEndpoint) bool {
		return item.ClusterID == req.GetClusterId() && item.EndpointID == req.GetPrivateLinkEndpoint().GetEndpointName()
	})
	if ok {
		return bizerr.New(
			bizerr.ErrCodeRecordExist,
			"endpoint %s already exists for cluster %d",
			req.GetPrivateLinkEndpoint().GetEndpointName(),
			req.GetClusterId(),
		)
	}

	return nil
}