package privatelink

import (
	"context"
	"strconv"

	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/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/workflow"
)

func (s *service) CreatePrivateLinkService(
	ctx context.Context,
	req *globalngv1.CreatePrivateLinkServiceRequest,
) (*entity.PrivateLinkService, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "privatelinkservice.CreatePrivateLinkService")
	defer span.End()

	if req.GetClusterId() == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}

	tidbGroupID := req.GetTidbGroupId()
	if tidbGroupID == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "tidb_group_id is required")
	}

	cluster, err := s.clusterRepo.Get(ctx, req.GetClusterId(), false)
	if err != nil {
		return nil, bizerr.WrapError(err, "get cluster failed")
	}
	if cluster == nil {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster not found")
	}
	if berr := servicecommon.CheckClusterResourcePoolLifecycle(
		ctx, s.resourcePoolRepo, cluster, resourcepoolpolicy.OperationUpdateNetwork,
	); berr != nil {
		return nil, berr
	}

	if cluster.State != globalngv1.Cluster_AVAILABLE {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster is not available")
	}

	_, err = s.tidbGroupRepo.Load(ctx, req.GetClusterId(), tidbGroupID)
	if err != nil {
		return nil, bizerr.WrapError(err, "load tidb group failed")
	}

	// Check if private link service already exists for this cluster and tidb group.
	existingServices, err := s.privatelinkRepo.ListPrivateLinkServices(ctx, &privatelink.Filter{
		ClusterID:   req.GetClusterId(),
		TidbGroupID: tidbGroupID,
	})
	if err != nil {
		return nil, bizerr.WrapError(err, "list private link services failed")
	}
	if len(existingServices) > 0 {
		return nil, bizerr.New(
			bizerr.ErrCodeRecordExist,
			"private link service already exists for cluster %d and tidb group %d",
			req.GetClusterId(),
			tidbGroupID,
		)
	}

	cloudRegion, err := s.regionRepo.Get(ctx, cluster.RegionId)
	if err != nil {
		return nil, bizerr.WrapError(err, "get region failed")
	}

	// private_link_connections is only honored for AWS imports (i.e. when
	// service_name is provided). Otherwise, the field is ignored per the
	// proto spec.
	importConnections := shouldImportPrivateLinkConnections(cloudRegion.CloudProviderId, req)
	if importConnections {
		if berr := s.validateImportPrivateLinkConnections(
			cloudRegion.CloudProviderId,
			tidbGroupID,
			req.GetPrivateLinkConnections(),
		); berr != nil {
			return nil, berr
		}
	}

	var out *entity.PrivateLinkService
	terr := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		if berr := servicecommon.LockClusterResourcePoolLifecycle(
			ctx, s.resourcePoolRepo, cluster, resourcepoolpolicy.OperationUpdateNetwork,
		); berr != nil {
			return berr
		}
		privateLinkServiceID := s.idGeneratorFacade.MustNew()
		privateLinkServiceEntity := &entity.PrivateLinkService{
			PrivateLinkServiceID: privateLinkServiceID,
			PrivateLinkService: &globalngv1.PrivateLinkService{
				ClusterId:         req.GetClusterId(),
				TidbGroupId:       tidbGroupID,
				ServicePort:       4000,
				RegionId:          cloudRegion.Id,
				RegionDisplayName: cloudRegion.DisplayName,
				CloudProviderId:   cloudRegion.CloudProviderId,
				ServiceName:       req.GetPrivateLinkService().GetServiceName(),
			},
		}

		if err := s.privatelinkRepo.CreatePrivateLinkService(ctx, privateLinkServiceEntity); err != nil {
			return bizerr.WrapError(err, "create private link service failed")
		}

		if importConnections {
			if err := s.createImportPrivateLinkConnections(
				ctx,
				cluster,
				cloudRegion,
				privateLinkServiceEntity,
				req.GetPrivateLinkConnections(),
			); err != nil {
				return err
			}
		}

		out = privateLinkServiceEntity
		out.State = globalngv1.PrivateLinkService_CREATING

		workflowErr := s.flow.RunDagWithBizId(
			ctx,
			common.CreatePrivateLinkService,
			strconv.FormatUint(cluster.OrganizationId, 10),
			"0",
			strconv.FormatUint(cluster.ClusterId, 10),
			map[string]string{
				workflow.KeyTraceId:             workflow.GetTraceId(ctx),
				common.KeyClusterId:             strconv.FormatUint(req.GetClusterId(), 10),
				common.KeyTidbGroupId:           strconv.FormatUint(tidbGroupID, 10),
				common.KeyTidbGroupIds:          strconv.FormatUint(tidbGroupID, 10),
				common.KeyPrivateLinkServiceIds: strconv.FormatUint(privateLinkServiceID, 10),
				common.KeyPrivateLbEnable:       "true",
			},
		)
		if workflowErr != nil {
			return bizerr.WrapErrorWithCode(workflowErr, bizerr.ErrCodeUnknown, "create async workflow failed")
		}

		return nil
	})
	if terr != nil {
		return nil, bizerr.WrapError(terr, "create private link service failed")
	}

	return out, nil
}

// shouldImportPrivateLinkConnections reports whether the supplied
// private_link_connections field on the request should be persisted. Per the
// proto spec, this is only the case for AWS when service_name is also set.
func shouldImportPrivateLinkConnections(
	cloudProvider globalcommonv1.CloudProvider_Name,
	req *globalngv1.CreatePrivateLinkServiceRequest,
) bool {
	if len(req.GetPrivateLinkConnections()) == 0 {
		return false
	}
	if cloudProvider != globalcommonv1.CloudProvider_AWS {
		return false
	}
	if req.GetPrivateLinkService().GetServiceName() == "" {
		return false
	}
	return true
}

// validateImportPrivateLinkConnections validates the connections supplied for
// the AWS import case. It reuses checkParams (per-connection field validation)
// and checkMaxConnectionPerGroup (group-level limit) from the
// CreatePrivateLinkConnection flow, and additionally enforces that endpoint
// ids are unique within the request.
func (s *service) validateImportPrivateLinkConnections(
	cloudProvider globalcommonv1.CloudProvider_Name,
	tidbGroupID uint64,
	conns []*globalngv1.PrivateLinkConnection,
) bizerr.BizError {
	seen := make(map[string]struct{}, len(conns))
	for _, c := range conns {
		if err := s.checkParams(c, cloudProvider); err != nil {
			return err
		}
		endpointID := c.GetEndpointId()
		if _, dup := seen[endpointID]; dup {
			return bizerr.New(bizerr.ErrCodeParams, "duplicated endpoint id: %q", endpointID)
		}
		seen[endpointID] = struct{}{}
	}

	// The private link service is brand new, so there are no existing
	// connections under (cluster_id, tidb_group_id); just ensure the import
	// batch does not exceed the per-group limit.
	if err := s.checkMaxConnectionPerGroup(cloudProvider, nil, tidbGroupID, len(conns)); err != nil {
		return err
	}
	return nil
}

// createImportPrivateLinkConnections persists the connections supplied by an
// AWS import request. Must be invoked inside the same transaction that
// creates the parent PrivateLinkService row.
func (s *service) createImportPrivateLinkConnections(
	ctx context.Context,
	cluster *entity.Cluster,
	region *entity.Region,
	psEntity *entity.PrivateLinkService,
	conns []*globalngv1.PrivateLinkConnection,
) bizerr.BizError {
	for _, c := range conns {
		connID := s.idGeneratorFacade.MustNew()
		connEntity := &entity.PrivateLinkConnection{
			PrivateLinkConnection: &globalngv1.PrivateLinkConnection{
				PrivateLinkConnectionId: strconv.FormatUint(connID, 10),
				ClusterId:               cluster.ClusterId,
				TidbGroupId:             psEntity.TidbGroupId,
				RegionId:                region.GetId(),
				RegionDisplayName:       region.GetDisplayName(),
				CloudProviderId:         region.GetCloudProviderId(),
				PrivateLinkService:      psEntity.GetServiceName(),
				EndpointType:            c.GetEndpointType(),
				EndpointId:              c.GetEndpointId(),
				PrivateIpAddress:        c.GetPrivateIpAddress(),
				PrivateDomainName:       c.GetPrivateDomainName(),
			},
			PrivateLinkServiceID: psEntity.PrivateLinkServiceID,
		}

		if err := s.privatelinkRepo.CreatePrivateLinkConnection(ctx, connEntity); err != nil {
			if err.ErrCode() == bizerr.ErrCodeRecordExist {
				return err.OverwriteMessage(
					"endpoint %s already exists for cluster %d",
					c.GetEndpointId(),
					cluster.ClusterId,
				)
			}
			return bizerr.WrapError(err, "create private link connection failed")
		}
	}
	return nil
}