package grpc

import (
	"context"
	"dataservices/commons/config"
	"dataservices/commons/logger"
	"fmt"
	"sync"
	"time"

	"github.com/avast/retry-go"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/tiinf/errs"
	tigrpc "github.com/tidbcloud/tiinf/transport/tigrpc/client"
	"go.uber.org/zap"
)

var (
	coreSvcOnce sync.Once
	coreCli     core.CoreServiceClient

	excludeClusterStatus = []core.ClusterStatus{
		core.ClusterStatus_CLUSTER_STATUS_DELETED,
		core.ClusterStatus_CLUSTER_STATUS_DELETECOMPLETE,
		core.ClusterStatus_CLUSTER_STATUS_CLEARED,
	}
)

const clusterBatchSize = 20
const orgBatchSize = 10

func coreRecordCounterMetric(err error) {
	recordRpcCounterMetric("core-svc", err)
}

func getCoreSvcClient() core.CoreServiceClient {
	coreSvcOnce.Do(func() {

		conn, err := tigrpc.NewClientConn(
			config.Global().GrpcConf.CoreSvcAddr,
			tigrpc.WithDefault(),
			tigrpc.WithLoadBalance(),
		)
		if err != nil {
			panic(fmt.Sprintf("Dial core-svc: %s", err))
		}

		coreCli = core.NewCoreServiceClient(conn)
	})

	return coreCli
}

func GetPrivateLinkService(ctx context.Context, req *core.GetPrivateLinkServiceReq) (resp *core.GetPrivateLinkServiceResp, err error) {

	defer func() { coreRecordCounterMetric(err) }()
	client := getCoreSvcClient()

	resp, err = client.GetPrivateLinkService(ctx, req)
	return resp, formatError(ctx, err, resp.GetBaseResp())
}

func ListClusters(ctx context.Context, req *core.ListClustersReq) (resp *core.ListClustersResp, err error) {

	defer func() { coreRecordCounterMetric(err) }()
	client := getCoreSvcClient()

	if req.Filter != nil && len(req.Filter.ExcludeStatuses) == 0 {
		req.Filter.ExcludeStatuses = excludeClusterStatus
	}
	retry.Do(func() error {
		newctx, cancel := context.WithTimeout(ctx, 5*time.Second)
		defer cancel()
		resp, err = client.ListClusters(newctx, req)
		return err
	}, retry.Attempts(3),
		retry.Delay(100*time.Millisecond),
		retry.DelayType(retry.BackOffDelay),
		retry.LastErrorOnly(true),
		retry.OnRetry(func(n uint, err error) {
			logger.Error(ctx, "failed to list cluster", zap.Uint("times", n), zap.Any("filter", req.Filter), zap.Error(err))
		}),
	)
	return resp, formatError(ctx, err, resp.GetBaseResp())
}

func ListClusterByIds(ctx context.Context, orgID, projectID uint64, clusterIDs []uint64) (clusterMap map[uint64]*core.Cluster, err error) {

	defer func() { coreRecordCounterMetric(err) }()

	resp, e := ListClusters(ctx, &core.ListClustersReq{
		Filter: &core.ClusterFilter{
			TenantIds:  []uint64{orgID},
			ProjectIds: []uint64{projectID},
			ClusterIds: clusterIDs,
		},
	})
	if e != nil {
		return nil, e
	}
	return resp.Clusters, nil
}

func GetCluster(ctx context.Context, tenantId, projectId, clusterId uint64) (*core.Cluster, error) {
	filter := &core.ClusterFilter{
		TenantIds:       []uint64{tenantId},
		ProjectIds:      []uint64{projectId},
		ClusterIds:      []uint64{clusterId},
		ExcludeStatuses: excludeClusterStatus,
	}
	listReq := &core.ListClustersReq{
		Filter:  filter,
		BaseReq: nil,
	}
	listResp, err := ListClusters(ctx, listReq)
	if err != nil {
		return nil, err
	}
	cluster, ok := listResp.Clusters[clusterId]
	if !ok {
		return nil, errs.NewNotFound("cluster is not found", nil)
	}
	return cluster, nil
}

func GetClusterWithoutPermCheck(ctx context.Context, clusterId uint64) (*core.Cluster, error) {
	filter := &core.ClusterFilter{
		ClusterIds:      []uint64{clusterId},
		ExcludeStatuses: excludeClusterStatus,
	}
	listReq := &core.ListClustersReq{
		Filter:  filter,
		BaseReq: nil,
	}
	listResp, err := ListClusters(ctx, listReq)
	if err != nil {
		return nil, err
	}
	cluster, ok := listResp.Clusters[clusterId]
	if !ok {
		return nil, errs.NewNotFound("cluster is not found", nil)
	}
	return cluster, nil
}

func GetClusterByID(ctx context.Context, clusterId uint64) (*core.Cluster, error) {
	filter := &core.ClusterFilter{
		ClusterIds: []uint64{clusterId},
		ExcludeStatuses: []core.ClusterStatus{
			core.ClusterStatus_CLUSTER_STATUS_UNSPECIFIED,
		},
	}
	listReq := &core.ListClustersReq{
		Filter:  filter,
		BaseReq: nil,
	}
	listResp, err := ListClusters(ctx, listReq)
	if err != nil {
		return nil, err
	}
	cluster, ok := listResp.Clusters[clusterId]
	if !ok {
		return nil, errs.NewNotFound("cluster is not found", nil)
	}
	return cluster, nil
}

func CreatePrivateLinkEndpoint(ctx context.Context, req *core.CreatePrivateLinkEndpointReq) (resp *core.CreatePrivateLinkEndpointResp, err error) {

	defer func() { coreRecordCounterMetric(err) }()
	client := getCoreSvcClient()

	resp, err = client.CreatePrivateLinkEndpoint(ctx, req)
	return resp, formatError(ctx, err, resp.GetBaseResp())
}

func ListClusterByIdsWPC(ctx context.Context, clusterIDs []uint64) (clusterMap map[uint64]*core.Cluster, err error) {

	defer func() { coreRecordCounterMetric(err) }()

	n := len(clusterIDs)
	clusterMap = make(map[uint64]*core.Cluster, len(clusterIDs))
	for i := 0; i < n; i += clusterBatchSize {
		j := i + clusterBatchSize
		if j > n {
			j = n
		}
		resp, e := ListClusters(ctx, &core.ListClustersReq{
			Filter: &core.ClusterFilter{
				ClusterIds: clusterIDs[i:j],
			},
		})
		if e != nil {
			err = e
			break
		}
		for k := range resp.Clusters {
			clusterMap[k] = resp.Clusters[k]
		}
	}
	return
}

func ListClusterByOrgs(ctx context.Context, orgIDs []uint64, isDevTier *bool, cloudProviderNames []string) (clusterMap map[uint64]*core.Cluster, err error) {

	defer func() { coreRecordCounterMetric(err) }()

	resp, err := ListClusters(ctx, &core.ListClustersReq{
		Filter: &core.ClusterFilter{
			TenantIds:          orgIDs,
			IsDevTier:          isDevTier,
			CloudProviderNames: cloudProviderNames,
		},
	})
	if err != nil {
		return nil, err
	}
	return resp.Clusters, nil
}