package grpcli

import (
	"context"
	"strings"
	"sync"

	"go.uber.org/zap"

	"github.com/pingcap/failpoint"
	"github.com/pingcap/log"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/pkgv2/bizerr"
	pkglog "github.com/tidbcloud/pkgv2/log"
)

var (
	DeletedClusterStatus = []core.ClusterStatus{
		core.ClusterStatus_CLUSTER_STATUS_DELETED, core.ClusterStatus_CLUSTER_STATUS_DELETECOMPLETE,
	}
	DeletedOrClearedClusterStatus = []core.ClusterStatus{
		core.ClusterStatus_CLUSTER_STATUS_DELETED, core.ClusterStatus_CLUSTER_STATUS_DELETECOMPLETE, core.ClusterStatus_CLUSTER_STATUS_CLEARED,
	}
	ReadyForBackupClusterStatus = []core.ClusterStatus{
		core.ClusterStatus_CLUSTER_STATUS_NORMAL,
	}
	RunningClusterStatus = []core.ClusterStatus{
		core.ClusterStatus_CLUSTER_STATUS_INITIALIZING, core.ClusterStatus_CLUSTER_STATUS_PREPARING,
		core.ClusterStatus_CLUSTER_STATUS_LAUNCHING, core.ClusterStatus_CLUSTER_STATUS_NORMAL,
		core.ClusterStatus_CLUSTER_STATUS_SCALING, core.ClusterStatus_CLUSTER_STATUS_RECOVERING,
	}
)

var (
	coreCli     core.CoreServiceClient
	coreSvcOnce sync.Once
)

func getCoreSvcClient() core.CoreServiceClient {
	coreSvcOnce.Do(func() {
		ctx := context.Background()
		addr := config.Global().GrpcConfig.CoreSvcAddr
		conn := grpcli.MustBuildGrpcClientConnection(ctx, addr)
		coreCli = core.NewCoreServiceClient(conn)
	})

	return coreCli
}

func ListClusterJobs(ctx context.Context, req *core.ListClusterJobsReq) (*core.ListClusterJobsResp, bizerr.BizError) {
	client := getCoreSvcClient()

	log.L().Info("list cluster job request",
		zap.Any("requestJob", req.Filter.JobType))
	resp, err := client.ListClusterJobs(ctx, req)
	if err != nil {
		return resp, bizerr.WrapGrpcError(err, "ListClusterJob")
	}

	log.L().Info("list cluster job response",
		zap.Any("count", len(resp.Jobs)))

	return resp, nil
}

func CreateClusterJob(ctx context.Context, req *core.CreateClusterJobReq) (*core.CreateClusterJobResp, bizerr.BizError) {
	client := getCoreSvcClient()

	log.L().Info("create cluster job request",
		zap.Any("request", req))
	resp, err := client.CreateClusterJob(ctx, req)
	if err != nil {
		return resp, bizerr.WrapGrpcError(err, "CreateClusterJob")
	}

	return resp, nil
}

func DeleteClusterJob(ctx context.Context, req *core.DeleteClusterJobReq) (*core.DeleteClusterJobResp, bizerr.BizError) {
	client := getCoreSvcClient()

	log.L().Info("delete cluster job request",
		zap.Any("request", req))
	resp, err := client.DeleteClusterJob(ctx, req)
	if err != nil {
		return resp, bizerr.WrapGrpcError(err, "DeleteClusterJob")
	}

	return resp, nil
}

func ListClustersById(ctx context.Context, clusterIds []uint64) (*core.ListClustersResp, bizerr.BizError) {
	clusterFilter := &core.ClusterFilter{
		ClusterIds:      clusterIds,
		ExcludeStatuses: nil,
	}
	listReq := &core.ListClustersReq{
		Filter:  clusterFilter,
		BaseReq: nil,
	}
	// TODO(ideascf): use ctx comming from parameter instead of context.Background()
	clustersResp, err := ListClusters(context.Background(), listReq)
	if err != nil {
		return nil, err
	}
	return clustersResp, nil
}

func GetClusterConfidentialInfo(ctx context.Context, req *core.GetClusterConfidentialInfoReq) (*core.GetClusterConfidentialInfoResp, bizerr.BizError) {
	client := getCoreSvcClient()

	resp, err := client.GetClusterConfidentialInfo(ctx, req)
	if err != nil {
		return resp, bizerr.WrapGrpcError(err, "GetClusterConfidentialInfo")
	}

	return resp, nil
}

// GetClusterConfidentialInfoByID gets cluster confidential information according to the cluster ID
// TODO: it will be removed later if the cloud_admin account is decoupled from the TiDB cluster
func GetClusterConfidentialInfoByID(ctx context.Context, clusterID uint64) (*core.ClusterConfidentialInfo, bizerr.BizError) {
	resp, bizErr := GetClusterConfidentialInfo(ctx,
		&core.GetClusterConfidentialInfoReq{
			ClusterId: clusterID,
			BaseReq:   nil,
		},
	)
	if bizErr != nil {
		return nil, bizErr
	}
	baseResp := resp.GetBaseResp()
	if baseResp.GetErrCode() != 0 {
		return nil, bizerr.Restore(bizerr.ErrCode(baseResp.GetErrCode()), baseResp.GetErrMsg())
	}

	return resp.GetInfo(), nil
}

// ListClustersReqOption is the option on chaning the behavior of ListClusters
type ListClustersReqOption func(req *core.ListClustersReq)

// ListClustersWithExcludeStatuses is the option to excludes the clusters with the specified statuses
func ListClustersWithExcludeStatuses(excludeStatuses []core.ClusterStatus) ListClustersReqOption {
	return func(req *core.ListClustersReq) {
		if len(excludeStatuses) <= 0 {
			return
		}
		if req.Filter == nil {
			req.Filter = new(core.ClusterFilter)
		}
		req.Filter.ExcludeStatuses = excludeStatuses
	}
}

// ListClustersWithIncludeStatuses is the option to includes the clusters with the specified statuses
func ListClustersWithIncludeStatuses(includeStatuses []core.ClusterStatus) ListClustersReqOption {
	return func(req *core.ListClustersReq) {
		if len(includeStatuses) <= 0 {
			return
		}
		if req.Filter == nil {
			req.Filter = new(core.ClusterFilter)
		}
		req.Filter.Statuses = includeStatuses
	}
}

// ListClustersWithDataPlaneInfo is the option to include data plan info in the cluster info
func ListClustersWithDataPlaneInfo() ListClustersReqOption {
	return func(req *core.ListClustersReq) {
		req.WithDataPlaneInfo = true
	}
}

func ListClusters(ctx context.Context, req *core.ListClustersReq) (*core.ListClustersResp, bizerr.BizError) {
	client := getCoreSvcClient()

	resp, e := client.ListClusters(ctx, req)
	if e != nil {
		return resp, bizerr.WrapGrpcError(e, "ListClusters")
	}
	if resp.BaseResp.ErrCode != 0 {
		return resp, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}
	pkglog.WithContextL(ctx).Debug("list cluster response", zap.Any("clusters", resp.Clusters))
	return resp, nil
}

// GetClusterByID gets the un-deleted cluster by default.
func GetClusterByID(ctx context.Context, clusterId uint64) (*core.Cluster, bizerr.BizError) {
	return GetClusterByIDAndExcludeStatuses(ctx, clusterId, DeletedOrClearedClusterStatus)
}

func GetAnyClusterByID(ctx context.Context, clusterId uint64) (*core.Cluster, bizerr.BizError) {
	return GetClusterByIDAndExcludeStatuses(ctx, clusterId, nil)
}

// GetClusterWithDataplaneByID gets the un-deleted cluster with dataplane by default.
func GetClusterWithDataplaneByID(ctx context.Context, clusterId uint64) (*core.Cluster, bizerr.BizError) {
	return GetClusterByIDGeneral(
		ctx,
		clusterId,
		ListClustersWithExcludeStatuses(DeletedOrClearedClusterStatus),
		ListClustersWithDataPlaneInfo(),
	)
}

func GetClusterByIDGeneral(ctx context.Context, clusterId uint64, opts ...ListClustersReqOption) (*core.Cluster, bizerr.BizError) {
	listReq := &core.ListClustersReq{
		Filter: &core.ClusterFilter{
			ClusterIds: []uint64{clusterId},
		},
		BaseReq: nil,
	}
	for _, o := range opts {
		o(listReq)
	}
	listResp, err := ListClusters(ctx, listReq)
	if err != nil {
		return nil, err
	}
	cluster, ok := listResp.Clusters[clusterId]
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster is not found,id=%d", clusterId)
	}
	return cluster, nil
}

// GetClusterByIDAndExcludeStatuses gets the cluster by cluster id and statuses.
func GetClusterByIDAndExcludeStatuses(ctx context.Context, clusterId uint64, excludeStatuses []core.ClusterStatus) (*core.Cluster, bizerr.BizError) {
	failpoint.Inject("MockGetClusterByIDAndStatusesStatus", func(v failpoint.Value) {
		cluster := core.Cluster{Id: clusterId, IsDevTier: false}
		cluster.Status = core.ClusterStatus(core.ClusterStatus_value[v.(string)])

		failpoint.Inject("MockGetClusterByIDAndStatusesStatusDevTier", func(_ failpoint.Value) {
			cluster.IsDevTier = true
		})

		failpoint.Inject("MockGetClusterByIDAndStatusesStatusToCleared", func(_ failpoint.Value) {
			cluster.Status = core.ClusterStatus_CLUSTER_STATUS_CLEARED
		})
		failpoint.Return(&cluster, nil)
	})

	return GetClusterByIDGeneral(ctx, clusterId, ListClustersWithExcludeStatuses(excludeStatuses))
}

// GetClusterByIDAndStatuses gets the cluster by cluster id and statuses.
func GetClusterByIDAndStatuses(ctx context.Context, clusterId uint64, statuses []core.ClusterStatus) (*core.Cluster, bizerr.BizError) {
	return GetClusterByIDGeneral(ctx, clusterId, ListClustersWithIncludeStatuses(statuses))
}

// Deprecated: use core-svc QueryCloudProviders instead.
func GetRegionsByProvider(ctx context.Context, provider string, deployType *core.DeployType) (map[string]*core.Region, error) {
	client := getCoreSvcClient()
	req := &core.ListCloudProvidersReq{
		FilterOutEmptyCloudProvider: true,
		FilterOutEmptyRegion:        true,
		FilterOutInactiveProvider:   true,
		FilterOutInactiveRegion:     true,
		DeployType:                  deployType,
	}
	resp, err := client.ListCloudProviders(ctx, req)
	if err != nil {
		return nil, err
	}

	var providerInfo *core.CloudProvider
	for name, cloudProvider := range resp.Name2CloudProvider {
		if strings.EqualFold(provider, name) {
			providerInfo = cloudProvider
		}
	}
	if providerInfo == nil {
		return map[string]*core.Region{}, nil
	}

	return providerInfo.Name2Region, nil
}

func ListNetworkPeersByCluster(ctx context.Context, cluster *core.Cluster) ([]*core.NetworkPeer, bizerr.BizError) {
	resp, err := coreCli.ListNetworkPeer(ctx,
		&core.ListNetworkPeerReq{
			Filter: &core.NetworkPeerFilter{
				CloudProvider: cluster.GetCloudProviderName(),
				RegionNames:   []string{cluster.GetRegionName()},
				Statuses:      []core.NetworkPeer_Status{core.NetworkPeer_STATUS_ACTIVE},
				ProjectIds:    []uint64{cluster.GetProjectId()},
			},
		},
	)
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "ListNetworkPeer")
	}
	baseResp := resp.GetBaseResp()
	if baseResp.GetErrCode() != 0 {
		return nil, bizerr.Restore(bizerr.ErrCode(baseResp.GetErrCode()), baseResp.GetErrMsg())
	}
	return resp.GetPeeringList(), nil
}