package controller

import (
	"context"
	"dataservices/commons/metrics"
	"dataservices/ddb"
	"dataservices/internal/entity"
	"dataservices/internal/repository_ddb"
	"fmt"
	"math/rand"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	config2 "dataservices/commons/config"
	dto "dataservices/internal/dto/dataapi_control"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
	"github.com/aws/aws-sdk-go-v2/service/ec2"
	"github.com/aws/aws-sdk-go-v2/service/ec2/types"
	"github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
)

var GlobalControlServer *ControlServer

type ControlServer struct {
	cancel   context.CancelFunc
	canceled atomic.Value

	awsCfg config2.AwsConfig
	vpcCfg config2.VpcConfig

	ddbCli *dynamodb.Client
	ec2Cli *ec2.Client

	createCh chan *VpcEndpointInfo
	deleteCh chan *VpcEndpointInfo

	mu            sync.Mutex
	concurrentMap map[Token]struct{}
}

func InitControlServer(ctx context.Context, awsCfg config2.AwsConfig, vpcCfg config2.VpcConfig) error {
	ctx, cancel := context.WithCancel(ctx)
	awsConfig := ddb.LoadAwsConfig(ctx, awsCfg.Region, "", "")
	ec2Cli := ec2.NewFromConfig(*awsConfig)
	server := &ControlServer{
		cancel:        cancel,
		awsCfg:        awsCfg,
		vpcCfg:        vpcCfg,
		ddbCli:        ddb.Client(),
		ec2Cli:        ec2Cli,
		createCh:      make(chan *VpcEndpointInfo, 128),
		deleteCh:      make(chan *VpcEndpointInfo, 128),
		concurrentMap: make(map[Token]struct{}),
	}

	pendingInfos, err := server.bootstrap(ctx)
	if err != nil {
		return err
	}

	go server.MaintainPrivateLinkLoop(ctx, pendingInfos)
	GlobalControlServer = server
	return nil
}

func (s *ControlServer) bootstrap(ctx context.Context) ([]entity.UserClusterInfo, error) {
	err := repository_ddb.CreateUserClusterInfoTableIfNotExist(ctx, s.ddbCli)
	if err != nil {
		return nil, err
	}
	// load pending private ddb from ddb.
	pendingInfos, err := repository_ddb.QueryUserClusterInfoWithPendingPrivateLink(ctx, s.ddbCli)
	if err != nil {
		return nil, err
	}
	tilog.Info(ctx, "load pending private link from ddb", zap.Int("pending-num", len(pendingInfos)))
	return pendingInfos, nil
}

func (s *ControlServer) Close() {
	if s.cancel != nil && s.canceled.CompareAndSwap(false, true) {
		s.cancel()
	}
}

func (s *ControlServer) MaintainPrivateLinkLoop(ctx context.Context, pendingInfos []entity.UserClusterInfo) {
	defer func() {
		if r := recover(); r != nil {
			tilog.Error(ctx, "MaintainPrivateLinkLoop panic", zap.Any("panic", r))
		}
		tilog.Info(ctx, "MaintainPrivateLinkLoop exit")
	}()

	cli := s.ec2Cli
	pendingPrivateLinks := make(map[string]*VpcEndpointInfo)
	needDeletedPrivateLinks := make(map[string]*VpcEndpointInfo)
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()

	for _, info := range pendingInfos {
		endpointInfo := &VpcEndpointInfo{
			id:          info.EndpointId,
			clusterInfo: info,
			begin:       time.Now(), // The timing is inaccurate.
		}
		pendingPrivateLinks[endpointInfo.id] = endpointInfo
	}
	for {
		select {
		case <-ctx.Done():
			return
		case info := <-s.createCh:
			pendingPrivateLinks[info.id] = info
			delete(needDeletedPrivateLinks, info.id)
			tilog.Info(ctx, "MaintainPrivateLinkLoop add pending private link",
				zap.Uint64("project_id", info.clusterInfo.ProjectId),
				zap.Uint64("cluster_id", info.clusterInfo.ClusterId),
				zap.String("endpoint-id", info.id),
				zap.Int("pending-private-link-num", len(pendingPrivateLinks)),
				zap.Int("need-deleted-private-link-num", len(needDeletedPrivateLinks)))
		case info := <-s.deleteCh:
			needDeletedPrivateLinks[info.id] = info
			delete(pendingPrivateLinks, info.id)
			tilog.Info(ctx, "MaintainPrivateLinkLoop need to delete private link",
				zap.Uint64("project_id", info.clusterInfo.ProjectId),
				zap.Uint64("cluster_id", info.clusterInfo.ClusterId),
				zap.String("endpoint-id", info.id),
				zap.Int("pending-private-link-num", len(pendingPrivateLinks)),
				zap.Int("need-deleted-private-link-num", len(needDeletedPrivateLinks)))
		case <-ticker.C:
		}
		for k, info := range pendingPrivateLinks {
			done, err := s.StepPrivateLink(ctx, cli, info)
			if err != nil {
				if config2.IsProd() {
					tilog.Error(ctx, "MaintainPrivateLinkLoop step pending private link failed.",
						zap.Uint64("project_id", info.clusterInfo.ProjectId),
						zap.Uint64("cluster_id", info.clusterInfo.ClusterId),
						zap.String("endpoint-id", info.id),
						zap.Error(err))
				} else {
					tilog.Warn(ctx, "MaintainPrivateLinkLoop step pending private link failed.",
						zap.Uint64("project_id", info.clusterInfo.ProjectId),
						zap.Uint64("cluster_id", info.clusterInfo.ClusterId),
						zap.String("endpoint-id", info.id),
						zap.Error(err))
				}
				continue
			}
			if done {
				delete(pendingPrivateLinks, k)
				metrics.WaitPrivateLinkAvailableDuration.Observe(time.Since(info.begin).Seconds())
			} else {
				if info.state != info.lastState || time.Since(info.lastLogTime) > time.Minute {
					info.lastLogTime = time.Now()
					tilog.Info(ctx, "MaintainPrivateLinkLoop step pending private link.",
						zap.Uint64("project_id", info.clusterInfo.ProjectId),
						zap.Uint64("cluster_id", info.clusterInfo.ClusterId),
						zap.String("endpoint-id", info.id),
						zap.String("state", info.state),
						zap.String("private-dns", info.clusterInfo.PrivateDNS),
						zap.String("endpoint-service", info.clusterInfo.EndpointService))
				}
				info.lastState = info.state
			}
		}

		for k, info := range needDeletedPrivateLinks {
			// delete the private dns name from ddb.
			var err error
			if info.isDeleteCluster {
				err = repository_ddb.DeleteUserClusterInfo(ctx, s.ddbCli, info.clusterInfo.ProjectId, info.clusterInfo.ClusterId)
			} else {
				info.clusterInfo.EndpointId = ""
				info.clusterInfo.PrivateDNS = ""
				err = repository_ddb.PutUserClusterInfo(ctx, s.ddbCli, &info.clusterInfo)
			}
			if err != nil {
				tilog.Error(ctx, "MaintainPrivateLinkLoop delete private dns info from ddb failed.",
					zap.Uint64("project_id", info.clusterInfo.ProjectId),
					zap.Uint64("cluster_id", info.clusterInfo.ClusterId),
					zap.String("endpoint-id", info.id),
					zap.Bool("is_delete_cluster", info.isDeleteCluster),
					zap.Error(err))
				continue
			}
			delete(needDeletedPrivateLinks, k)
			tilog.Info(ctx, "MaintainPrivateLinkLoop deleted private link.",
				zap.Uint64("project_id", info.clusterInfo.ProjectId),
				zap.Uint64("cluster_id", info.clusterInfo.ClusterId),
				zap.String("endpoint-id", info.id),
				zap.Bool("is_delete_cluster", info.isDeleteCluster),
				zap.Int("need-deleted-private-link-num", len(needDeletedPrivateLinks)))
		}
	}
}

func (s *ControlServer) StepPrivateLink(ctx context.Context, cli *ec2.Client, info *VpcEndpointInfo) (bool, error) {
	err := info.DescribeVpcEndpoint(ctx, cli)
	if err != nil {
		return false, err
	}
	if !info.available {
		// wait util vpc endpoint available.
		return false, nil
	}
	if !info.privateDnsEnabled {
		// Enable private link dns.
		err = EnablePrivateLinkDNS(ctx, cli, info.id)
		tilog.Info(ctx, "PrivateLink dns is enabled.",
			zap.Uint64("project_id", info.clusterInfo.ProjectId),
			zap.Uint64("cluster_id", info.clusterInfo.ClusterId),
			zap.String("endpoint-id", info.id),
			zap.String("state", info.state),
			zap.String("private-dns", info.clusterInfo.PrivateDNS),
			zap.String("endpoint-service", info.clusterInfo.EndpointService),
			zap.Bool("success", err == nil))
		return false, err
	}
	if info.clusterInfo.PrivateDNS == "" {
		return false, nil
	}
	// store the private dns name into ddb.
	err = repository_ddb.PutUserClusterInfo(ctx, s.ddbCli, &info.clusterInfo)
	if err != nil {
		return false, err
	}
	tilog.Info(ctx, "PrivateLink is available now",
		zap.Uint64("project_id", info.clusterInfo.ProjectId),
		zap.Uint64("cluster_id", info.clusterInfo.ClusterId),
		zap.Duration("wait_time", time.Since(info.begin)),
		zap.String("endpoint-id", info.id),
		zap.String("state", info.state),
		zap.String("private-dns", info.clusterInfo.PrivateDNS),
		zap.String("endpoint-service", info.clusterInfo.EndpointService))
	return true, nil
}

func (s *ControlServer) CreatePrivateLink(ctx context.Context, projectId, clusterId uint64, serviceName string) (*dto.CreatePrivateLinkResp, errs.Error) {
	token := s.GetToken(projectId, clusterId)
	if token == nil {
		// token is nil indicate already having a private link in creating.
		return nil, errs.NewBadRequest("already in creating", nil)
	}
	defer s.ReleaseToken(token)

	clusterInfo, _ := repository_ddb.GetUserClusterInfo(ctx, s.ddbCli, projectId, clusterId)
	if clusterInfo == nil {
		return nil, errs.NewBadRequest("cluster not found", nil)
	}
	if clusterInfo.EndpointId != "" && clusterInfo.PrivateDNS != "" {
		// already created
		return &dto.CreatePrivateLinkResp{
			BaseResp: dto.BaseResp{
				Code:    http.StatusOK,
				Message: "ok",
			},
			VpcEndpointId: clusterInfo.EndpointId,
			PrivateDNS:    clusterInfo.PrivateDNS,
		}, nil
	}

	if clusterInfo.EndpointId == "" {
		cli := s.ec2Cli
		subnetIds, err := s.GetEndpointServicesAzSubnet(ctx, cli, serviceName)
		if err != nil {
			return nil, errs.NewInternalServerError(err.Error(), err)
		}
		resp, err := CreateVpcEndpoint(ctx, cli, clusterId, s.vpcCfg.VpcId, s.vpcCfg.SecurityGroupId, serviceName, subnetIds)
		if err != nil {
			return nil, errs.NewInternalServerError(err.Error(), err)
		}
		clusterInfo.EndpointId = *resp.VpcEndpoint.VpcEndpointId
		clusterInfo.EndpointService = serviceName
		// store endpoint info into ddb.
		err = repository_ddb.PutUserClusterInfo(ctx, s.ddbCli, clusterInfo)
		if err != nil {
			// only log error but not return err here. since this won't affect correctness.
			tilog.Error(ctx, "CreatePrivateLink store user cluster info into ddb failed", zap.Uint64("cluster_id", clusterId), zap.String("vpc_id", s.vpcCfg.VpcId), zap.Stringp("vpce_id", resp.VpcEndpoint.VpcEndpointId), zap.Error(err))
		}
	}

	info := &VpcEndpointInfo{
		id:          clusterInfo.EndpointId,
		clusterInfo: *clusterInfo,
		begin:       time.Now(),
	}
	s.createCh <- info
	return &dto.CreatePrivateLinkResp{
		BaseResp: dto.BaseResp{
			Code:    http.StatusOK,
			Message: "ok",
		},
		VpcEndpointId: clusterInfo.EndpointId,
	}, nil
}

func (s *ControlServer) GetEndpointServicesAzSubnet(ctx context.Context, cli *ec2.Client, serviceName string) ([]string, error) {
	resp, err := cli.DescribeVpcEndpointServices(ctx, &ec2.DescribeVpcEndpointServicesInput{
		ServiceNames: []string{serviceName},
	})
	if err != nil || len(resp.ServiceDetails) == 0 {
		return nil, err
	}
	az2SubnetMap := s.vpcCfg.GetAz2SubnetMap()
	subnetIds := []string{}
	for _, az := range resp.ServiceDetails[0].AvailabilityZones {
		subnetId := az2SubnetMap[az]
		if subnetId != "" {
			subnetIds = append(subnetIds, subnetId)
		}
	}
	if len(subnetIds) == 0 {
		tilog.Error(ctx, "can't find available zone in config for the endpoint service",
			zap.String("service-name", serviceName),
			zap.Strings("service-availability-zones", resp.ServiceDetails[0].AvailabilityZones),
			zap.Any("az-config", az2SubnetMap))
		return nil, fmt.Errorf("can't find available zone in config for the endpoint service")
	}
	if len(subnetIds) <= 1 {
		return subnetIds, nil
	}
	// use random az
	subnetId := subnetIds[rand.Intn(len(subnetIds))]
	return []string{subnetId}, nil
}

func (s *ControlServer) DeletePrivateLink(ctx context.Context, projectId, clusterId uint64, isDeleteCluster bool) (*dto.DeletePrivateLinkResp, errs.Error) {
	token := s.GetToken(projectId, clusterId)
	if token == nil {
		// token is nil indicate already having a private link in process.
		return nil, errs.NewBadRequest("already in process", nil)
	}
	defer s.ReleaseToken(token)
	cli := s.ec2Cli

	// If it has been deleted, return directly
	clusterInfo, err := repository_ddb.GetUserClusterInfo(ctx, s.ddbCli, projectId, clusterId)
	if err != nil {
		if errs.IsError(err, errs.ErrNotFound) {
			return &dto.DeletePrivateLinkResp{
				BaseResp: dto.BaseResp{
					Code:    http.StatusOK,
					Message: "ok",
				},
			}, nil
		}
		return nil, errs.NewBadRequest("cluster not found", err)
	}
	if clusterInfo.EndpointId == "" {
		s.deleteCh <- &VpcEndpointInfo{
			id:              clusterInfo.EndpointId,
			clusterInfo:     *clusterInfo,
			begin:           time.Now(),
			isDeleteCluster: isDeleteCluster,
		}
		return &dto.DeletePrivateLinkResp{
			BaseResp: dto.BaseResp{
				Code:    http.StatusOK,
				Message: "ok",
			},
		}, nil
	}

	endpointID := clusterInfo.EndpointId
	resp, err := cli.DescribeVpcEndpoints(ctx, &ec2.DescribeVpcEndpointsInput{
		VpcEndpointIds: []string{endpointID},
	})
	if err != nil {
		if strings.Contains(err.Error(), "InvalidVpcEndpointId.NotFound") {
			tilog.Warn(ctx, "DescribeVpcEndpoints not found endpoint, mab be already deleted", zap.Uint64("cluster_id", clusterId), zap.String("endpoint-id", endpointID), zap.Error(err))
		} else {
			tilog.Error(ctx, "DescribeVpcEndpoints failed", zap.Uint64("cluster_id", clusterId), zap.String("endpoint-id", endpointID), zap.Error(err))
			return nil, errs.NewInternalServerError("failed to describe vpc endpoints", err)
		}
	} else if len(resp.VpcEndpoints) == 0 {
		tilog.Warn(ctx, "DescribeVpcEndpoints got empty result, mab be already deleted", zap.Uint64("cluster_id", clusterId), zap.String("endpoint-id", endpointID))
	} else {
		resp, err := DeleteVpcEndpoint(ctx, cli, clusterInfo.EndpointId)
		if err != nil {
			return nil, errs.NewInternalServerError(err.Error(), err)
		}
		if len(resp.Unsuccessful) > 0 {
			msg := fmt.Sprintf("DeleteVpcEndpoint failed, error: %v", *resp.Unsuccessful[0].Error.Message)
			return nil, errs.NewInternalServerError(msg, nil)
		}
	}

	s.deleteCh <- &VpcEndpointInfo{
		id:              clusterInfo.EndpointId,
		clusterInfo:     *clusterInfo,
		begin:           time.Now(),
		isDeleteCluster: isDeleteCluster,
	}
	return &dto.DeletePrivateLinkResp{
		BaseResp: dto.BaseResp{
			Code:    http.StatusOK,
			Message: "ok",
		},
		VpcEndpointId: clusterInfo.EndpointId,
	}, nil
}

func CreateVpcEndpoint(ctx context.Context, cli *ec2.Client, clusterId uint64, vpcId, securityGroupId, serviceName string, subnetIds []string) (resp *ec2.CreateVpcEndpointOutput, err error) {
	startTime := time.Now()
	defer func() {
		metrics.CreateVpcEndpointDuration.WithLabelValues(strconv.FormatBool(err == nil)).Observe(time.Since(startTime).Seconds())
	}()

	name := fmt.Sprintf("regional-control-endpoint-cluster-%v", clusterId)
	resp, err = cli.CreateVpcEndpoint(ctx, &ec2.CreateVpcEndpointInput{
		ServiceName:      aws.String(serviceName),
		VpcId:            aws.String(vpcId),
		SubnetIds:        subnetIds,
		SecurityGroupIds: []string{securityGroupId},
		VpcEndpointType:  types.VpcEndpointTypeInterface,
		TagSpecifications: []types.TagSpecification{
			{
				ResourceType: types.ResourceTypeVpcEndpoint,
				Tags: []types.Tag{
					{
						Key:   aws.String("Name"),
						Value: aws.String(name),
					},
					{
						Key:   aws.String("tenant"),
						Value: aws.String("tcds"),
					},
				},
			},
		},
	})
	if err != nil {
		tilog.Error(ctx, "CreateVpcEndpoint failed.", zap.Error(err))
		return nil, err
	}
	if resp.VpcEndpoint == nil || resp.VpcEndpoint.VpcEndpointId == nil || len(*resp.VpcEndpoint.VpcEndpointId) == 0 {
		tilog.Error(ctx, "CreateVpcEndpoint return empty vpc endpoint id.")
		return nil, errs.NewInternalServerError("CreateVpcEndpoint return empty vpc endpoint id.", nil)
	}
	endpointId := *resp.VpcEndpoint.VpcEndpointId
	tilog.Info(ctx, "CreateVpcEndpoint success.", zap.String("vpc-id", endpointId), zap.Strings("subnet-ids", subnetIds), zap.String("service-name", serviceName))
	return resp, nil
}

type VpcEndpointInfo struct {
	id              string
	clusterInfo     entity.UserClusterInfo
	begin           time.Time
	isDeleteCluster bool

	state             string
	available         bool
	privateDnsEnabled bool

	lastState   string
	lastLogTime time.Time
}

func (info *VpcEndpointInfo) DescribeVpcEndpoint(ctx context.Context, cli *ec2.Client) error {
	resp, err := cli.DescribeVpcEndpoints(ctx, &ec2.DescribeVpcEndpointsInput{
		VpcEndpointIds: []string{info.id},
	})
	if err != nil {
		if config2.IsProd() {
			tilog.Error(ctx, "DescribeVpcEndpoints failed", zap.String("endpoint-id", info.id), zap.Error(err))
		} else {
			tilog.Info(ctx, "DescribeVpcEndpoints failed", zap.String("endpoint-id", info.id), zap.Error(err))
		}
		return err
	}
	for _, e := range resp.VpcEndpoints {
		if *e.VpcEndpointId != info.id {
			tilog.Error(ctx, "DescribeVpcEndpoints return unexpected endpoint",
				zap.String("expected-endpoint-id", info.id),
				zap.String("got-endpoint-id", *e.VpcEndpointId))
			continue
		}
		info.state = fmt.Sprintf("%v", e.State)
		info.available = strings.ToLower(info.state) == "available"
		for _, dns := range e.DnsEntries {
			if strings.Contains(*dns.DnsName, "privatelink") {
				info.clusterInfo.PrivateDNS = *dns.DnsName
			}
		}
		if e.PrivateDnsEnabled != nil {
			info.privateDnsEnabled = *e.PrivateDnsEnabled
		}
		return nil
	}
	tilog.Error(ctx, "DescribeVpcEndpoints not found endpoint", zap.String("endpoint-id", info.id))
	return fmt.Errorf("Vpc endpoint id %v not found", info.id)
}

func WaitPrivateLinkAvailable(ctx context.Context, cli *ec2.Client, endpointId string, timeout time.Duration) error {
	// wait vpc endpoint available(It should be accepted by endpoint service).
	start := time.Now()
	logTime := start
	for {
		if time.Since(start) > timeout {
			tilog.Error(ctx, "WaitPrivateLinkAvailable timeout", zap.String("endpoint-id", endpointId))
			return fmt.Errorf("Wait vpc endpoint id %v available timeout", endpointId)
		}
		resp, err := cli.DescribeVpcEndpoints(ctx, &ec2.DescribeVpcEndpointsInput{
			VpcEndpointIds: []string{endpointId},
		})
		if err != nil {
			tilog.Error(ctx, "DescribeVpcEndpoints failed", zap.String("endpoint-id", endpointId), zap.Error(err))
			return err
		}
		if len(resp.VpcEndpoints) == 0 {
			tilog.Error(ctx, "DescribeVpcEndpoints not found endpoint", zap.String("endpoint-id", endpointId))
			return fmt.Errorf("Vpc endpoint id %v not found", endpointId)
		}
		available := false
		state := ""
		for _, e := range resp.VpcEndpoints {
			state = strings.ToLower(fmt.Sprintf("%v", e.State))
			available = state == "available"
		}
		if time.Since(logTime).Seconds() > 2 || available {
			logTime = time.Now()
			tilog.Info(ctx, "DescribeVpcEndpoints success, wait until available", zap.String("endpoint-id", endpointId), zap.Bool("available", available), zap.String("state", state))
		}
		if available {
			break
		}
		time.Sleep(time.Millisecond * 500)
	}
	return nil
}

func EnablePrivateLinkDNS(ctx context.Context, cli *ec2.Client, endpointId string) error {
	// enable private dns.
	resp, err := cli.ModifyVpcEndpoint(ctx, &ec2.ModifyVpcEndpointInput{
		VpcEndpointId:     aws.String(endpointId),
		PrivateDnsEnabled: aws.Bool(true),
	})
	if err != nil {
		tilog.Error(ctx, "ModifyVpcEndpoint failed", zap.String("endpoint-id", endpointId), zap.Error(err))
		return err
	}
	tilog.Info(ctx, "ModifyVpcEndpoint success", zap.String("endpoint-id", endpointId), zap.Bool("return", *resp.Return))
	return nil
}

func GetPrivateLinkDNS(ctx context.Context, cli *ec2.Client, endpointId string) (string, error) {
	resp, err := cli.DescribeVpcEndpoints(ctx, &ec2.DescribeVpcEndpointsInput{
		VpcEndpointIds: []string{endpointId},
	})
	if err != nil {
		return "", err
	}
	for _, e := range resp.VpcEndpoints {
		for _, dns := range e.DnsEntries {
			if strings.Contains(*dns.DnsName, "privatelink") {
				return *dns.DnsName, nil
			}
		}
	}
	return "", nil
}

func DeleteVpcEndpoint(ctx context.Context, cli *ec2.Client, endpointId string) (*ec2.DeleteVpcEndpointsOutput, error) {
	resp, err := cli.DeleteVpcEndpoints(ctx, &ec2.DeleteVpcEndpointsInput{
		VpcEndpointIds: []string{endpointId},
	})
	if err != nil {
		return nil, err
	}
	return resp, nil
}

func DescribeSubnets(ctx context.Context, cli *ec2.Client, vpcId string) (*ec2.DescribeSubnetsOutput, error) {
	resp, err := cli.DescribeSubnets(ctx, &ec2.DescribeSubnetsInput{
		Filters: []types.Filter{
			{
				Name:   aws.String("vpc-id"),
				Values: []string{vpcId},
			},
		},
	})
	if err != nil {
		tilog.Error(ctx, "DescribeSubnets failed", zap.String("vpc-id", vpcId), zap.Error(err))
		return nil, err
	}
	return resp, nil
}

type Token struct {
	projectId uint64
	clusterId uint64
}

func (s *ControlServer) GetToken(projectId, clusterId uint64) *Token {
	token := Token{projectId, clusterId}
	s.mu.Lock()
	defer s.mu.Unlock()
	_, exist := s.concurrentMap[token]
	if exist {
		return nil
	}
	s.concurrentMap[token] = struct{}{}
	return &token
}

func (s *ControlServer) ReleaseToken(token *Token) {
	if token == nil {
		return
	}
	s.mu.Lock()
	delete(s.concurrentMap, *token)
	s.mu.Unlock()
}