package grpcli

import (
	"context"

	"github.com/samber/lo"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	globalv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	serverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	v1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	tigrpc "github.com/tidbcloud/tiinf/transport/tigrpc/client"
	"go.uber.org/zap"
	"google.golang.org/grpc"
)

var DeletedClusterStatus = []v1beta1.ClusterState{
	v1beta1.ClusterState_DELETED, v1beta1.ClusterState_DELETING,
}

type MgmtSvcGrpcClient interface {
	GetServerlessRegional(ctx context.Context, clusterId uint64) (*serverlessv1.ServerlessRegionalServiceClusterInfo, bizerr.BizError)

	UpdateServerlessRegionalAnnotations(ctx context.Context, clusterId uint64, annotations map[string]string) bizerr.BizError

	// GetClusterByID returns the cluster with full view if it's not in DELETED/DELETING state, otherwise returns not found error.
	GetClusterByID(ctx context.Context, clusterID uint64) (*mgmt.Cluster, bizerr.BizError)

	// GetClusterBasicByID is similar to GetClusterByID but returns the cluster with basic view
	GetClusterBasicByID(ctx context.Context, clusterID uint64) (*mgmt.Cluster, bizerr.BizError)

	// GetAnyClusterByID returns the cluster regardless of its state (including DELETING/DELETED).
	GetAnyClusterByID(ctx context.Context, clusterID uint64) (*mgmt.Cluster, bizerr.BizError)

	// GetAnyDedicatedClusterBasicByID fetches a dedicated cluster by ID using CLUSTER_VIEW_BASIC, which is served
	// from the tidb-management-service database and does not require a regional K8s API call.
	// Use this instead of GetAnyClusterByID in code paths that must remain available during
	// CRRF / disaster-recovery scenarios where the regional control plane may be unavailable.
	GetAnyDedicatedClusterBasicByID(ctx context.Context, clusterID uint64) (*mgmt.Cluster, bizerr.BizError)

	// GetDedicatedClusterBasicByID returns the dedicated cluster with BASIC view for any non-deleted state.
	// Returns ErrCodeRecordNotFound if the cluster does not exist or is deleted.
	// Use this in place of coreCli.GetAvailableClusterByID.
	GetDedicatedClusterBasicByID(ctx context.Context, clusterID uint64) (*mgmt.Cluster, bizerr.BizError)

	RestoreDedicatedCluster(ctx context.Context, req *mgmt.RestoreDedicatedClusterRequest) (*mgmt.Cluster, bizerr.BizError)

	ListClusters(ctx context.Context, in *mgmt.ListClustersRequest, opts ...grpc.CallOption) (*mgmt.ListClustersResponse, bizerr.BizError)

	ListCustomerManagedEncryptionKey(ctx context.Context, req *mgmt.ListCustomerManagedEncryptionKeysRequest) (*mgmt.ListCustomerManagedEncryptionKeysResponse, bizerr.BizError)

	SetClusterCustomStatus(ctx context.Context, in *mgmt.SetClusterCustomStatusRequest, opts ...grpc.CallOption) (*mgmt.ClusterCustomStatus, bizerr.BizError)

	DeleteClusterCustomStatus(ctx context.Context, in *mgmt.DeleteClusterCustomStatusRequest, opts ...grpc.CallOption) bizerr.BizError

	IsClusterCustomStatusExisted(ctx context.Context, in *mgmt.IsClusterCustomStatusExistedRequest, opts ...grpc.CallOption) (*mgmt.IsClusterCustomStatusExistedResponse, bizerr.BizError)

	GetResourcePoolByID(ctx context.Context, resourcePoolID uint64) (*globalngv1.ResourcePool, bizerr.BizError)

	GetNextGenDatabaseAuditLogSetting(ctx context.Context, clusterID uint64) (*globalngv1.AuditLogSetting, bizerr.BizError)
}

type mgmtSvcGrpcClient struct {
	MgmtSvcGrpcClient
	addr                string
	client              mgmt.ClusterServiceClient
	cmekCli             mgmt.CmekServiceClient
	regionClient        mgmt.RegionServiceClient
	clusterStatusClient mgmt.ClusterCustomStatusServiceClient
	resourcePoolClient  mgmt.ResourcePoolServiceClient
	auditLogClient      globalngv1.DatabaseAuditLogServiceClient
}

func NewMgmtSvcGrpcClient(ctx context.Context, addr string) MgmtSvcGrpcClient {
	logger := log.GetLogger(ctx)
	conn, err := tigrpc.NewClientConn(addr, tigrpc.WithDefault())
	logger.Info("Initializing mgmt service client", zap.Any("err", err), zap.String("addr", addr))
	if err != nil {
		logger.Fatal("initialize client failed", zap.String("addr", addr), zap.Error(err))
	}
	return &mgmtSvcGrpcClient{
		addr:                addr,
		client:              mgmt.NewClusterServiceClient(conn),
		cmekCli:             mgmt.NewCmekServiceClient(conn),
		regionClient:        mgmt.NewRegionServiceClient(conn),
		clusterStatusClient: mgmt.NewClusterCustomStatusServiceClient(conn),
		resourcePoolClient:  mgmt.NewResourcePoolServiceClient(conn),
		auditLogClient:      globalngv1.NewDatabaseAuditLogServiceClient(conn),
	}
}

func (m *mgmtSvcGrpcClient) GetServerlessRegional(ctx context.Context, clusterId uint64) (*serverlessv1.ServerlessRegionalServiceClusterInfo, bizerr.BizError) {
	req := &mgmt.ListClustersRequest{
		Filter: &mgmt.ClusterFilter{
			ClusterIds:    []uint64{clusterId},
			ExcludeStates: DeletedClusterStatus,
			ClusterType:   mgmt.DeployType_DEPLOY_TYPE_SERVERLESS.Enum(),
		},
	}
	listResp, err := m.client.ListClusters(ctx, req)
	if err != nil {
		return nil, bizerr.WrapError(err, "GetServerlessClusterFullView failed")
	}
	if len(listResp.Clusters) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster %d is not found", clusterId)
	}
	if listResp.Clusters[0].GetServerlessCluster().GetCluster() == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "cluster %d regional info is empty", clusterId)

	}
	return listResp.Clusters[0].GetServerlessCluster().GetCluster(), nil
}

func (m *mgmtSvcGrpcClient) GetClusterByID(ctx context.Context, clusterID uint64) (*mgmt.Cluster, bizerr.BizError) {
	req := &mgmt.ListClustersRequest{
		Filter: &mgmt.ClusterFilter{
			ClusterIds:    []uint64{clusterID},
			ExcludeStates: DeletedClusterStatus,
			View:          mgmt.ClusterFilter_CLUSTER_VIEW_FULL,
		},
	}
	listResp, err := m.client.ListClusters(ctx, req)
	if err != nil {
		return nil, bizerr.WrapError(err, "GetClusterByID failed")
	}
	if len(listResp.Clusters) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster %d is not found", clusterID)
	}
	return listResp.Clusters[0], nil
}

func (m *mgmtSvcGrpcClient) GetClusterBasicByID(ctx context.Context, clusterID uint64) (*mgmt.Cluster, bizerr.BizError) {
	req := &mgmt.ListClustersRequest{
		Filter: &mgmt.ClusterFilter{
			ClusterIds:    []uint64{clusterID},
			ExcludeStates: DeletedClusterStatus,
			View:          mgmt.ClusterFilter_CLUSTER_VIEW_BASIC,
		},
	}
	listResp, err := m.ListClusters(ctx, req)
	if err != nil {
		return nil, bizerr.WrapError(err, "GetClusterBasicByID failed")
	}
	if len(listResp.Clusters) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster %d is not found", clusterID)
	}
	return listResp.Clusters[0], nil
}

func (m *mgmtSvcGrpcClient) GetAnyClusterByID(ctx context.Context, clusterID uint64) (*mgmt.Cluster, bizerr.BizError) {
	req := &mgmt.ListClustersRequest{
		Filter: &mgmt.ClusterFilter{
			ClusterIds: []uint64{clusterID},
			View:       mgmt.ClusterFilter_CLUSTER_VIEW_FULL,
		},
	}
	listResp, err := m.client.ListClusters(ctx, req)
	if err != nil {
		return nil, bizerr.WrapError(err, "GetAnyClusterByID failed")
	}
	if len(listResp.Clusters) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster %d is not found", clusterID)
	}
	return listResp.Clusters[0], nil
}

func (m *mgmtSvcGrpcClient) GetAnyDedicatedClusterBasicByID(ctx context.Context, clusterID uint64) (*mgmt.Cluster, bizerr.BizError) {
	req := &mgmt.ListClustersRequest{
		Filter: &mgmt.ClusterFilter{
			ClusterIds:  []uint64{clusterID},
			View:        mgmt.ClusterFilter_CLUSTER_VIEW_BASIC,
			ClusterType: mgmt.DeployType_DEPLOY_TYPE_DEDICATED.Enum(),
		},
	}
	listResp, err := m.ListClusters(ctx, req)
	if err != nil {
		return nil, bizerr.WrapError(err, "GetAnyDedicatedClusterBasicByID failed")
	}
	if len(listResp.Clusters) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster %d is not found", clusterID)
	}
	return listResp.Clusters[0], nil
}

func (m *mgmtSvcGrpcClient) GetDedicatedClusterBasicByID(ctx context.Context, clusterID uint64) (*mgmt.Cluster, bizerr.BizError) {
	notDeleted := false
	req := &mgmt.ListClustersRequest{
		Filter: &mgmt.ClusterFilter{
			ClusterIds:  []uint64{clusterID},
			View:        mgmt.ClusterFilter_CLUSTER_VIEW_BASIC,
			Deleted:     &notDeleted,
			ClusterType: mgmt.DeployType_DEPLOY_TYPE_DEDICATED.Enum(),
		},
	}
	listResp, err := m.ListClusters(ctx, req)
	if err != nil {
		return nil, bizerr.WrapError(err, "GetDedicatedClusterBasicByID failed")
	}
	if len(listResp.GetClusters()) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster %d is not found or unavailable", clusterID)
	}
	return listResp.GetClusters()[0], nil
}

func (m *mgmtSvcGrpcClient) UpdateServerlessRegionalAnnotations(ctx context.Context, clusterId uint64, annotations map[string]string) bizerr.BizError {
	_, err := m.client.UpdateServerlessRegionalClusterAnnotations(ctx, &mgmt.UpdateServerlessRegionalClusterAnnotationsRequest{
		ClusterId:   clusterId,
		Annotations: annotations,
	})
	if err != nil {
		return bizerr.WrapError(err, "UpdateServerlessRegionalAnnotations failed")
	}
	return nil
}

func (m *mgmtSvcGrpcClient) RestoreDedicatedCluster(ctx context.Context, req *mgmt.RestoreDedicatedClusterRequest) (*mgmt.Cluster, bizerr.BizError) {
	cluster, err := m.client.RestoreDedicatedCluster(ctx, req)
	if err != nil {
		return nil, bizerr.WrapError(err, "RestoreDedicatedCluster failed")
	}
	return cluster, nil
}

func (m *mgmtSvcGrpcClient) ListClusters(ctx context.Context, in *mgmt.ListClustersRequest, opts ...grpc.CallOption) (*mgmt.ListClustersResponse, bizerr.BizError) {
	if in != nil && in.Filter != nil &&
		in.Filter.View == mgmt.ClusterFilter_CLUSTER_VIEW_BASIC &&
		in.Filter.AllowNonFullView == nil {
		// It's a workaround that the mgmt-service always returns full view even if given basic view.
		in.Filter.AllowNonFullView = lo.ToPtr(true)
	}

	clusters, err := m.client.ListClusters(ctx, in, opts...)
	if err != nil {
		return nil, bizerr.WrapError(err, "List clusters failed")
	}
	return clusters, nil
}

func (m *mgmtSvcGrpcClient) ListCustomerManagedEncryptionKey(ctx context.Context, req *mgmt.ListCustomerManagedEncryptionKeysRequest) (*mgmt.ListCustomerManagedEncryptionKeysResponse, bizerr.BizError) {
	keys, err := m.cmekCli.ListCustomerManagedEncryptionKeys(ctx, req)
	if err != nil {
		return nil, bizerr.WrapError(err, "List customer managed encryption keys failed")
	}
	return keys, nil
}

func (m *mgmtSvcGrpcClient) SetClusterCustomStatus(ctx context.Context, in *mgmt.SetClusterCustomStatusRequest, opts ...grpc.CallOption) (*mgmt.ClusterCustomStatus, bizerr.BizError) {
	status, err := m.clusterStatusClient.SetClusterCustomStatus(ctx, in, opts...)
	if err != nil {
		return nil, bizerr.WrapError(err, "SetClusterCustomStatus failed")
	}
	return status, nil
}

func (m *mgmtSvcGrpcClient) DeleteClusterCustomStatus(ctx context.Context, in *mgmt.DeleteClusterCustomStatusRequest, opts ...grpc.CallOption) bizerr.BizError {
	_, err := m.clusterStatusClient.DeleteClusterCustomStatus(ctx, in, opts...)
	if err != nil {
		return bizerr.WrapError(err, "DeleteClusterCustomStatus failed")
	}
	return nil
}

func (m *mgmtSvcGrpcClient) IsClusterCustomStatusExisted(ctx context.Context, in *mgmt.IsClusterCustomStatusExistedRequest, opts ...grpc.CallOption) (*mgmt.IsClusterCustomStatusExistedResponse, bizerr.BizError) {
	status, err := m.clusterStatusClient.IsClusterCustomStatusExisted(ctx, in, opts...)
	if err != nil {
		return nil, bizerr.WrapError(err, "IsClusterCustomStatusExisted failed")
	}
	return status, nil
}

func (m *mgmtSvcGrpcClient) GetResourcePoolByID(ctx context.Context, resourcePoolID uint64) (*globalngv1.ResourcePool, bizerr.BizError) {
	req := &globalngv1.GetResourcePoolRequest{
		ResourcePoolId: resourcePoolID,
	}
	resp, err := m.resourcePoolClient.GetResourcePool(ctx, req)
	if err != nil {
		if bizerr.ParseGormErr2code(err) == bizerr.ErrCodeRecordNotFound {
			return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "resource pool %d is not found", resourcePoolID)
		}

		return nil, bizerr.WrapError(err, "GetResourcePoolByID failed")
	}
	return resp.ResourcePool, nil
}

// DedicatedGlobalGrpcClient wraps cluster config backup RPCs on the dedicated global ClusterService.
type DedicatedGlobalGrpcClient interface {
	EnableClusterConfigBackup(ctx context.Context, req *globalv1.EnableClusterConfigBackupRequest) (*globalv1.ClusterConfigBackupPolicy, bizerr.BizError)
	DisableClusterConfigBackup(ctx context.Context, req *globalv1.DisableClusterConfigBackupRequest) (*globalv1.ClusterConfigBackupPolicy, bizerr.BizError)
	QueryCidrs(ctx context.Context, req *globalv1.QueryCidrsRequest) (*globalv1.QueryCidrsResponse, bizerr.BizError)
}

type dedicatedGlobalGrpcClient struct {
	client    globalv1.ClusterServiceClient
	vpcClient globalv1.VpcServiceClient
}

func NewDedicatedGlobalGrpcClient(ctx context.Context, addr string) DedicatedGlobalGrpcClient {
	logger := log.GetLogger(ctx)
	logger.Info("initializing dedicated global grpc client", zap.String("addr", addr))
	conn := MustBuildGrpcClientConnection(ctx, addr)
	return &dedicatedGlobalGrpcClient{
		client:    globalv1.NewClusterServiceClient(conn),
		vpcClient: globalv1.NewVpcServiceClient(conn),
	}
}

func (c *dedicatedGlobalGrpcClient) EnableClusterConfigBackup(ctx context.Context, req *globalv1.EnableClusterConfigBackupRequest) (*globalv1.ClusterConfigBackupPolicy, bizerr.BizError) {
	resp, err := c.client.EnableClusterConfigBackup(ctx, req)
	if err != nil {
		return nil, globalServiceError2BizError(err)
	}
	return resp, nil
}

func (c *dedicatedGlobalGrpcClient) DisableClusterConfigBackup(ctx context.Context, req *globalv1.DisableClusterConfigBackupRequest) (*globalv1.ClusterConfigBackupPolicy, bizerr.BizError) {
	resp, err := c.client.DisableClusterConfigBackup(ctx, req)
	if err != nil {
		return nil, globalServiceError2BizError(err)
	}
	return resp, nil
}

func (c *dedicatedGlobalGrpcClient) QueryCidrs(ctx context.Context, req *globalv1.QueryCidrsRequest) (*globalv1.QueryCidrsResponse, bizerr.BizError) {
	resp, err := c.vpcClient.QueryCidrs(ctx, req)
	if err != nil {
		return nil, globalServiceError2BizError(err)
	}
	return resp, nil
}

func (m *mgmtSvcGrpcClient) GetNextGenDatabaseAuditLogSetting(ctx context.Context, clusterID uint64) (*globalngv1.AuditLogSetting, bizerr.BizError) {
	resp, err := m.auditLogClient.GetAuditLogSetting(ctx, &globalngv1.GetAuditLogSettingRequest{
		ClusterId: clusterID,
	})
	if err != nil {
		return nil, globalServiceError2BizError(err)
	}
	return resp, nil
}