package grpcli

import (
	"context"
	"maps"
	"strconv"

	"github.com/elliotchance/pie/v2"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.buf.build/protocolbuffers/go/linuxgit/tidbcloud/base"
	"go.uber.org/zap"
)

type CoreGrpcClient interface {
	GetClient() core.CoreServiceClient

	GetAnyClusterByID(ctx context.Context, clusterId uint64) (*core.Cluster, bizerr.BizError)
	GetAvailableClusterByID(ctx context.Context, clusterId uint64) (*core.Cluster, bizerr.BizError)
	ListClusters(ctx context.Context, req *core.ListClustersReq) (*core.ListClustersResp, bizerr.BizError)
	ListClustersWithoutIds(ctx context.Context, req *core.ListClustersReq) (*core.ListClustersResp, bizerr.BizError)
	CountClusters(ctx context.Context, req *core.CountClustersReq) (int64, bizerr.BizError)

	ListCloudRegions(ctx context.Context, req *core.ListCloudRegionsReq) (map[string]map[string]*core.CloudRegion, bizerr.BizError)
	GetClusterConfidentialInfo(ctx context.Context, req *core.GetClusterConfidentialInfoReq) (*core.GetClusterConfidentialInfoResp, bizerr.BizError)

	CreateClusterJob(ctx context.Context, req *core.CreateClusterJobReq) (*core.CreateClusterJobResp, bizerr.BizError)
	DeleteClusterJob(ctx context.Context, req *core.DeleteClusterJobReq) (*core.DeleteClusterJobResp, bizerr.BizError)
	ListClusterJobs(ctx context.Context, req *core.ListClusterJobsReq) (*core.ListClusterJobsResp, bizerr.BizError)

	GetAwsCmek(ctx context.Context, req *core.GetAwsCmekReq) (*core.GetAwsCmekResp, bizerr.BizError)

	IsClusterJobRegistered(ctx context.Context, req *core.IsClusterJobRegisteredReq) (*core.IsClusterJobRegisteredResp, bizerr.BizError)
	CreateClusterSnapshot(ctx context.Context, clusterID uint64) (uint64, bizerr.BizError)
	GetClusterSnapshot(ctx context.Context, clusterId uint64, snapshotId uint64) (*core.ClusterSnapshot, bizerr.BizError)
	ListClusterSnapshot(ctx context.Context, tenantId, projectId uint64, snapshotIds []uint64) (map[uint64]*core.ClusterSnapshot, bizerr.BizError)

	GetProjectRegionCIDR(
		ctx context.Context, projectID uint64, cloudProvider, region string) (*core.Cidr, bizerr.BizError)

	GetClusterAccessPoints(ctx context.Context, clusterId uint64) ([]*core.AccessPoint, bizerr.BizError)
}

var DeletedOrClearedClusterStatus = []core.ClusterStatus{
	core.ClusterStatus_CLUSTER_STATUS_DELETED, core.ClusterStatus_CLUSTER_STATUS_DELETECOMPLETE, core.ClusterStatus_CLUSTER_STATUS_CLEARED,
}

var DefaultPageSize int64 = 100

type CoreGrpcClientImpl struct {
	addr   string
	Client core.CoreServiceClient
}

func NewCoreGrpcClient(ctx context.Context, addr string) CoreGrpcClient {
	logger := log.GetLogger(ctx)
	logger.Info("initializing core-svc client", zap.String("addr", addr))

	conn := MustBuildGrpcClientConnection(ctx, addr)
	coreCli := core.NewCoreServiceClient(conn)
	return &CoreGrpcClientImpl{
		addr:   addr,
		Client: coreCli,
	}
}

func (c *CoreGrpcClientImpl) GetClient() core.CoreServiceClient {
	return c.Client
}

func (c *CoreGrpcClientImpl) IsClusterJobRegistered(ctx context.Context, req *core.IsClusterJobRegisteredReq) (*core.IsClusterJobRegisteredResp, bizerr.BizError) {
	resp, err := c.Client.IsClusterJobRegistered(ctx, req)
	if err != nil {
		return resp, bizerr.WrapGrpcError(err, "IsClusterJobRegistered")
	}

	return resp, nil
}

func (c *CoreGrpcClientImpl) ListClustersWithoutIds(
	ctx context.Context, req *core.ListClustersReq,
) (*core.ListClustersResp, bizerr.BizError) {
	if req.Pager != nil {
		return c.ListClusters(ctx, req)
	}

	var (
		page              int64 = 1
		respClustersCount       = DefaultPageSize
		clusters                = make(map[uint64]*core.Cluster)
		resp              *core.ListClustersResp
		err               bizerr.BizError
	)
	for respClustersCount == DefaultPageSize {
		req.Pager = &base.Pager{
			Page:     page,
			PageSize: DefaultPageSize,
		}
		resp, err = c.ListClusters(ctx, req)
		if err != nil {
			return resp, bizerr.WrapGrpcError(err, "ListClustersWithoutIds Page: "+strconv.FormatInt(page, 10))
		}
		if len(resp.Clusters) != 0 {
			maps.Copy(clusters, resp.Clusters)

		}
		respClustersCount = int64(len(resp.Clusters))
		page += 1
	}
	resp.Clusters = clusters
	return resp, nil
}

func (c *CoreGrpcClientImpl) ListClusters(
	ctx context.Context, req *core.ListClustersReq,
) (*core.ListClustersResp, bizerr.BizError) {
	resp, e := c.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)
	}
	return resp, nil
}

func (c *CoreGrpcClientImpl) CountClusters(
	ctx context.Context, req *core.CountClustersReq,
) (int64, bizerr.BizError) {
	resp, e := c.Client.CountClusters(ctx, req)
	if e != nil {
		return 0, bizerr.WrapGrpcError(e, "CountClusters")
	}
	if resp.BaseResp.ErrCode != 0 {
		return 0, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}
	return resp.Count, nil
}

func (c *CoreGrpcClientImpl) ListClusterByIDsAndExcludeStatuses(ctx context.Context, clusterIds []uint64, excludeStatuses []core.ClusterStatus) (*core.ListClustersResp, bizerr.BizError) {
	filter := &core.ClusterFilter{
		ClusterIds:      clusterIds,
		ExcludeStatuses: excludeStatuses,
	}
	listReq := &core.ListClustersReq{
		Filter:  filter,
		BaseReq: nil,
	}
	listResp, err := c.ListClusters(ctx, listReq)
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "ListClusters")
	}
	if listResp.BaseResp.ErrCode != 0 {
		return nil, bizerr.Restore(bizerr.ErrCode(listResp.BaseResp.ErrCode), listResp.BaseResp.ErrMsg)
	}
	return listResp, nil
}

func (c *CoreGrpcClientImpl) GetClusterByIDAndExcludeStatuses(ctx context.Context, clusterId uint64, excludeStatuses []core.ClusterStatus) (*core.Cluster, bizerr.BizError) {
	listResp, bizError := c.ListClusterByIDsAndExcludeStatuses(ctx, []uint64{clusterId}, excludeStatuses)
	if bizError != nil {
		return nil, bizError
	}
	cluster, ok := listResp.Clusters[clusterId]
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster id=%d is not found", clusterId)
	}
	return cluster, nil
}

func (c *CoreGrpcClientImpl) GetAvailableClusterByID(ctx context.Context, clusterId uint64) (*core.Cluster, bizerr.BizError) {
	return c.GetClusterByIDAndExcludeStatuses(ctx, clusterId, DeletedOrClearedClusterStatus)
}

func (c *CoreGrpcClientImpl) ListCloudRegions(ctx context.Context, req *core.ListCloudRegionsReq) (map[string]map[string]*core.CloudRegion, bizerr.BizError) {
	resp, err := c.Client.ListCloudRegions(ctx, req)
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "ListCloudRegions")
	}
	if resp.BaseResp.ErrCode != 0 {
		return nil, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}
	cloudRegionMap := make(map[string]map[string]*core.CloudRegion)
	pie.Of(resp.Regions).Each(func(cloudRegion *core.CloudRegion) {
		if _, ok := cloudRegionMap[cloudRegion.CloudProviderName]; !ok {
			cloudRegionMap[cloudRegion.CloudProviderName] = make(map[string]*core.CloudRegion)
		}
		cloudRegionMap[cloudRegion.CloudProviderName][cloudRegion.Name] = cloudRegion
	})
	return cloudRegionMap, nil
}

func (c *CoreGrpcClientImpl) GetClusterConfidentialInfo(ctx context.Context, req *core.GetClusterConfidentialInfoReq) (*core.GetClusterConfidentialInfoResp, bizerr.BizError) {
	resp, err := c.Client.GetClusterConfidentialInfo(ctx, req)
	if err != nil {
		return resp, bizerr.WrapGrpcError(err, "GetClusterConfidentialInfo")
	}
	return resp, nil
}

func (c *CoreGrpcClientImpl) CreateClusterJob(ctx context.Context, req *core.CreateClusterJobReq) (*core.CreateClusterJobResp, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	logger.Info("create cluster job request",
		zap.Any("request", req))

	resp, err := c.Client.CreateClusterJob(ctx, req)
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "CreateClusterJob")
	}
	if resp.BaseResp.ErrCode != 0 {
		return nil, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}
	return resp, nil
}

func (c *CoreGrpcClientImpl) DeleteClusterJob(ctx context.Context, req *core.DeleteClusterJobReq) (*core.DeleteClusterJobResp, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	logger.Info("delete cluster job request",
		zap.Any("request", req))

	resp, err := c.Client.DeleteClusterJob(ctx, req)
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "DeleteClusterJob")
	}
	if resp.BaseResp.ErrCode != 0 {
		return nil, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}
	return resp, nil
}

func (c *CoreGrpcClientImpl) ListClusterJobs(ctx context.Context, req *core.ListClusterJobsReq) (*core.ListClusterJobsResp, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	logger.Info("list cluster job request",
		zap.Any("requestJob", req.Filter.JobType))
	resp, err := c.Client.ListClusterJobs(ctx, req)
	if err != nil {
		return resp, bizerr.WrapGrpcError(err, "ListClusterJob")
	}
	if resp.BaseResp.ErrCode != 0 {
		return nil, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}

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

	return resp, nil
}

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

func (c *CoreGrpcClientImpl) GetAwsCmek(ctx context.Context, req *core.GetAwsCmekReq) (*core.GetAwsCmekResp, bizerr.BizError) {
	resp, err := c.Client.GetAwsCmek(ctx, req)
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "GetAwsCmek")
	}
	return resp, nil
}

func IsClusterDeleted(status core.ClusterStatus) bool {
	for _, deletedStatus := range DeletedOrClearedClusterStatus {
		if status == deletedStatus {
			return true
		}
	}
	return false
}

func (c *CoreGrpcClientImpl) CreateClusterSnapshot(ctx context.Context, clusterID uint64) (uint64, bizerr.BizError) {
	req := &core.CreateClusterSnapshotReq{
		ClusterId: clusterID,
	}
	resp, e := c.Client.CreateClusterSnapshot(ctx, req)
	if e != nil {
		return 0, bizerr.WrapGrpcError(e, "CreateClusterSnapshot")
	}
	if resp.BaseResp.ErrCode != 0 {
		return 0, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}
	return resp.SnapshotId, nil
}

func (c *CoreGrpcClientImpl) GetClusterSnapshot(
	ctx context.Context, clusterId uint64, snapshotId uint64,
) (*core.ClusterSnapshot, bizerr.BizError) {
	req := &core.ListClusterSnapshotsReq{
		Filter: &core.ClusterSnapshotFilter{
			ClusterIds:  []uint64{clusterId},
			SnapshotIds: []uint64{snapshotId},
		},
	}
	resp, err := c.Client.ListClusterSnapshots(ctx, req)
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "ListClusterSnapshot")
	}
	if resp.BaseResp.ErrCode != 0 {
		return nil, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}
	if len(resp.Snapshots) == 0 {
		return nil, bizerr_wrapper.NewNotFoundErr("snapshot %d not found", snapshotId)
	}
	return resp.Snapshots[0], nil
}

func (c *CoreGrpcClientImpl) ListClusterSnapshot(
	ctx context.Context, tenantId, projectId uint64, snapshotIds []uint64,
) (map[uint64]*core.ClusterSnapshot, bizerr.BizError) {
	req := &core.ListClusterSnapshotsReq{
		Filter: &core.ClusterSnapshotFilter{
			TenantIds:   []uint64{tenantId},
			ProjectIds:  []uint64{projectId},
			SnapshotIds: snapshotIds,
		},
	}
	resp, err := c.Client.ListClusterSnapshots(ctx, req)
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "ListClusterSnapshot")
	}
	if resp.BaseResp.ErrCode != 0 {
		return nil, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}

	snapshotMap := make(map[uint64]*core.ClusterSnapshot)
	for _, snapshot := range resp.Snapshots {
		snapshotMap[snapshot.Id] = snapshot
	}
	return snapshotMap, nil
}

func (c *CoreGrpcClientImpl) GetProjectRegionCIDR(
	ctx context.Context, projectID uint64, cloudProvider, region string,
) (*core.Cidr, bizerr.BizError) {
	regionCIDRs, err := c.ListProjectCIDRs(ctx, projectID, cloudProvider)
	if err != nil {
		return nil, err
	}
	for _, regionCIDR := range regionCIDRs {
		if regionCIDR.RegionName == region && regionCIDR.CloudProviderName == cloudProvider {
			return regionCIDR, nil
		}
	}
	return nil, bizerr.New(bizerr.ErrCodeRecordNotFound,
		"the region(%s) cidr is not found, cidrs of (projectId=%d, cloudProvider=%s) are %+v",
		region, projectID, cloudProvider, regionCIDRs)
}

func (c *CoreGrpcClientImpl) ListProjectCIDRs(
	ctx context.Context, projectID uint64, cloudProvider string,
) (regionCIDRs []*core.Cidr, bizErr bizerr.BizError) {
	resp, err := c.Client.QueryCidrs(ctx, &core.QueryCidrsRequest{
		Filter: &core.QueryCidrsFilter{ProjectIds: []uint64{projectID}, CloudProviderName: cloudProvider},
	})
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "QueryCidrs")
	}
	if resp.BaseResp.ErrCode != 0 {
		return nil, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}

	return resp.Items, nil
}

func (c *CoreGrpcClientImpl) GetClusterAccessPoints(
	ctx context.Context, clusterId uint64,
) ([]*core.AccessPoint, bizerr.BizError) {
	resp, err := c.Client.QueryAccessPoints(ctx, &core.QueryAccessPointsReq{
		Filter: &core.QueryAccessPointsFilter{ClusterIds: []uint64{clusterId}},
		View:   core.AccessPointView_ACCESS_POINT_VIEW_FULL,
	})
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "QueryAccessPoints error, clusterId: %d", clusterId)
	}

	if len(resp.AccessPoints) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound,
			"QueryAccessPoints error, clusterId: %d, no AccessPoints found", clusterId)
	}

	return resp.AccessPoints, nil
}