package coreportalapi

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	httptransport "github.com/go-openapi/runtime/client"
	"github.com/go-openapi/strfmt"
	"github.com/go-sql-driver/mysql"
	"github.com/samber/lo"
	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"
	gormmysql "gorm.io/driver/mysql"
	"gorm.io/gorm"

	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/coreportalapi/client"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/coreportalapi/client/cluster_service"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/coreportalapi/model"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/sql"
)

type Cli struct {
	*client.GRPCForOurDBaaSCentralService

	su *suite.Suite
}

func NewClient(su *suite.Suite, host, bearerToken string) (*Cli, error) {
	trans := httptransport.New(host, "", []string{"https"})
	// Apply the bearer token
	trans.DefaultAuthentication = httptransport.BearerToken(bearerToken)
	return &Cli{su: su, GRPCForOurDBaaSCentralService: client.New(trans, strfmt.Default)}, nil
}

func (cli *Cli) RespJson(resp any) string {
	b, err := json.Marshal(resp)
	require.NoErrorf(cli.su.T(), err, "marshal error: %s", err)
	return string(b)
}

func (cli *Cli) ListClustersAllowBizErr(
	orgID, prjID string,
	page, perPage int64,
	isServerless *bool,
) *model.CentralClusters {
	params := cluster_service.NewListClustersParams()
	if orgID != "" {
		params.SetOrgID(orgID)
	}
	if prjID != "" {
		params.SetProjectID(prjID)
	}
	params.SetPage(page)
	params.SetPerPage(perPage)

	if isServerless != nil {
		params.SetFilterIsServerless(isServerless)
	}

	resp, err := cli.ClusterService.ListClusters(params, nil)
	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))

	return resp.Payload
}

func (cli *Cli) ListClusters(orgID, prjID string, page, perPage int64) []*model.CentralCluster {
	resp := cli.ListClustersAllowBizErr(orgID, prjID, page, perPage, nil)
	require.Falsef(
		cli.su.T(),
		resp.BaseResp != nil && resp.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)
	return resp.Items
}

func (cli *Cli) ListProviders(orgID, prjID string, isDevTier bool) []*model.CentralCloudProvider {
	params := cluster_service.NewListCloudProvidersV2Params()
	params.SetOrgID(orgID)
	params.SetProjectID(prjID)
	params.SetBody(cluster_service.ListCloudProvidersV2Body{
		IsDevTier: lo.ToPtr(isDevTier),
	})

	resp, err := cli.ClusterService.ListCloudProvidersV2(params, nil)

	require.NoError(cli.su.T(), err)
	require.Falsef(
		cli.su.T(),
		resp.Payload == nil || (resp.Payload.BaseResp != nil && resp.Payload.BaseResp.ErrCode != "0"),
		"invalid response: %s",
		cli.RespJson(resp),
	)

	require.GreaterOrEqualf(
		cli.su.T(),
		len(resp.Payload.CloudProviders),
		1,
		"length cloud providers response less then 1: %s",
		resp.String(),
	)

	return resp.Payload.CloudProviders
}

// Project cidr must be set prior to creating a cluster, manually。
func (cli *Cli) CheckProjectRegionCidr(orgID, prjID, provider, region string) {
	params := cluster_service.NewGetProjectCIDRParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjID)
	params.SetProvider(provider)
	params.SetRegion(region)

	resp, err := cli.ClusterService.GetProjectCIDR(params, nil)

	require.NoError(cli.su.T(), err)
	require.NotNilf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))

	require.NotEmptyf(cli.su.T(), *resp.Payload.Cidr, "cidr not found: %s", resp.String())
}

func (cli *Cli) GenerateClusterName(orgID, prjID string) *string {
	params := cluster_service.NewGenClusterNameParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjID)

	resp, err := cli.ClusterService.GenClusterName(params, nil)

	require.NoError(cli.su.T(), err)
	require.Falsef(
		cli.su.T(),
		resp.Payload == nil || (resp.Payload.BaseResp != nil && resp.Payload.BaseResp.ErrCode != "0"),
		"invalid response: %s",
		cli.RespJson(resp),
	)

	require.NotEmptyf(cli.su.T(), *resp.Payload.Name, "cluster name not found: %s", resp.String())

	return resp.Payload.Name
}

func (cli *Cli) QueryNodeProfiles(
	orgID, prjID, provider, region string,
	isDevTier bool,
) []*model.CentralComponentNodeProfiles {
	params := cluster_service.NewCalcNodeProfilesV2Params()
	params.SetOrgID(orgID)
	params.SetProjectID(prjID)
	params.SetBody(cluster_service.CalcNodeProfilesV2Body{
		IsDevTier: lo.ToPtr(isDevTier),
		Provider:  lo.ToPtr(provider),
		Region:    lo.ToPtr(region),
	})

	resp, err := cli.ClusterService.CalcNodeProfilesV2(params, nil)

	require.NoError(cli.su.T(), err)
	require.Falsef(
		cli.su.T(),
		resp.Payload == nil || (resp.Payload.BaseResp != nil && resp.Payload.BaseResp.ErrCode != "0"),
		"invalid response: %s",
		cli.RespJson(resp),
	)

	require.GreaterOrEqualf(
		cli.su.T(),
		len(resp.Payload.NodeProfiles),
		3,
		"length node profiles response less then 3, should contain tidb/tikv/tiflash: %s",
		resp.String(),
	)

	return resp.Payload.NodeProfiles
}

func (cli *Cli) EstimateClusterPrice(
	orgID, prjID, provider, region string,
	components []*model.CentralClusterComponent,
) []*model.CentralCalcClusterFeeRespComponentCost {
	params := cluster_service.NewCalcClusterFeeParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjID)
	params.SetBody(cluster_service.CalcClusterFeeBody{
		IsDevTier:  lo.ToPtr(false),
		Provider:   lo.ToPtr(provider),
		Region:     lo.ToPtr(region),
		Components: components,
	})

	resp, err := cli.ClusterService.CalcClusterFee(params, nil)

	require.NoError(cli.su.T(), err)
	require.Falsef(
		cli.su.T(),
		resp.Payload == nil || (resp.Payload.BaseResp != nil && resp.Payload.BaseResp.ErrCode != "0"),
		"invalid response: %s",
		cli.RespJson(resp),
	)

	require.GreaterOrEqualf(
		cli.su.T(),
		len(resp.Payload.Costs),
		2,
		"length costs response less then 2: %s",
		resp.String(),
	)

	return resp.Payload.Costs
}

func (cli *Cli) CreateClusterAllowBizErr(
	orgID, prjID, provider, region, name string,
	port int32,
	components []*model.CentralClusterComponent,
	isDevTier bool,
	spendLimit int64,
) *model.CentralCreateClusterResp {
	time.Sleep(1 * time.Second)
	params := cluster_service.NewCreateClusterV2Params()
	params.SetOrgID(orgID)
	params.SetProjectID(prjID)
	params.SetBody(cluster_service.CreateClusterV2Body{
		IsDevTier:  lo.ToPtr(isDevTier),
		Name:       lo.ToPtr(name),
		Port:       lo.ToPtr(port),
		Provider:   lo.ToPtr(provider),
		Region:     lo.ToPtr(region),
		Components: components,
		SpendLimit: spendLimit,
	})

	resp, err := cli.ClusterService.CreateClusterV2(params, nil)

	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))

	return resp.Payload
}

func (cli *Cli) CreateCidr(orgID, prjectID, cidr, provider, region string) {
	params := cluster_service.NewCreateCidrParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjectID)

	var cloudProvider *model.CentralCloudRegionCloudProvider
	switch provider {
	case "aws":
		cloudProvider = lo.ToPtr(model.CentralCloudRegionCloudProviderAWS)
	case "gcp":
		cloudProvider = lo.ToPtr(model.CentralCloudRegionCloudProviderGCP)
	}
	require.NotNil(cli.su.T(), cloudProvider)

	params.SetBody(cluster_service.CreateCidrBody{
		Cidr: lo.ToPtr(cidr),
		Region: &model.CentralCloudRegion{
			Name:     region,
			Provider: cloudProvider,
		},
		ValidateOnly: false,
	})
	resp, err := cli.ClusterService.CreateCidr(params, nil)
	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))
	require.Falsef(
		cli.su.T(),
		resp.Payload.BaseResp != nil && resp.Payload.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)
}

func (cli *Cli) CreateCluster(
	orgID, prjID, provider, region, name string,
	port int32,
	components []*model.CentralClusterComponent,
	isDevTier bool,
	spendLimit int64,
) *string {
	resp := cli.CreateClusterAllowBizErr(orgID, prjID, provider, region, name, port, components, isDevTier, spendLimit)

	require.Falsef(
		cli.su.T(),
		resp.BaseResp != nil && resp.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)
	require.NotEmptyf(cli.su.T(), *resp.ID, "cluster id not found: %s", cli.RespJson(resp))

	return resp.ID
}

func (cli *Cli) PollClusterStatus(orgID, prjID, clusterID string) *model.CentralClusterStatus {
	clusters := cli.ListClusters(orgID, prjID, 1, 100)

	cluster, ok := lo.Find(clusters, func(c *model.CentralCluster) bool {
		return c != nil && c.ID != nil && *c.ID == clusterID
	})

	require.Truef(cli.su.T(), ok, "cluster not found: %s", clusterID)

	require.NotEmptyf(cli.su.T(), *cluster.Status, "cluster status not found: %s", clusterID)

	return cluster.Status
}

func (cli *Cli) GetClusterWithError(orgID, prjID, clusterID string) error {
	params := cluster_service.NewGetClusterParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjID)
	params.SetID(clusterID)

	_, err := cli.ClusterService.GetCluster(params, nil)
	return err
}

func (cli *Cli) GetClusterAllowBizErr(orgID, prjID, clusterID string) *model.CentralGetClusterResp {
	params := cluster_service.NewGetClusterParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjID)
	params.SetID(clusterID)

	resp, err := cli.ClusterService.GetCluster(params, nil)

	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))

	return resp.Payload
}

func (cli *Cli) GetCluster(orgID, prjID, clusterID string) *model.CentralCluster {
	resp := cli.GetClusterAllowBizErr(orgID, prjID, clusterID)

	require.Falsef(
		cli.su.T(),
		resp.BaseResp != nil && resp.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)
	return resp.Cluster
}

func (cli *Cli) SetPasswordAllowBizErr(orgID, prjID, clusterID, password string) *model.CentralDefaultResp {
	params := cluster_service.NewChangePasswordParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjID)
	params.SetClusterID(clusterID)
	params.SetBody(cluster_service.ChangePasswordBody{
		Password: lo.ToPtr(password),
	})

	resp, err := cli.ClusterService.ChangePassword(params, nil)

	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))

	return resp.Payload
}

func (cli *Cli) SetPassword(orgID, prjID, clusterID, password string) {
	resp := cli.SetPasswordAllowBizErr(orgID, prjID, clusterID, password)

	require.Falsef(
		cli.su.T(),
		resp.BaseResp != nil && resp.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)
}

func (cli *Cli) SetPublicInternetAccess(
	orgID, prjID, clusterID string,
) (*cluster_service.UpdateTrafficFilterOK, error) {
	params := cluster_service.NewUpdateTrafficFilterParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjID)
	params.SetClusterID(clusterID)
	params.SetBody(cluster_service.UpdateTrafficFilterBody{
		TrafficFilters: &model.CentralTrafficFilterList{
			Cidrs: []*model.CentralTrafficFilter{
				{
					Cidr:        lo.ToPtr("0.0.0.0/0"),
					Description: lo.ToPtr("Allow Access from Anywhere."),
				},
			},
		},
	})

	return cli.ClusterService.UpdateTrafficFilter(params, nil)
}

func (cli *Cli) ConnectDatabase(host, port, user, password, database, tlsConfig string) *gorm.DB {
	config := mysql.NewConfig()
	config.Net = "tcp"
	config.User = user
	config.Addr = fmt.Sprintf("%s:%s", host, port)
	config.Passwd = password
	config.DBName = database
	config.ParseTime = true
	config.Timeout = 30 * time.Second
	config.ReadTimeout = 30 * time.Second
	config.WriteTimeout = 30 * time.Second
	config.Params = map[string]string{"charset": "utf8mb4"}
	config.Collation = "utf8mb4_bin"

	// Only allow TLS config key to be one of the following values.
	// Currently, not support customized server CA or enable mTLS (server verify client).
	require.Containsf(
		cli.su.T(),
		[]string{"", "true", "false", "skip-verify", "preferred"},
		tlsConfig,
		"tls mod not supported: %s, refer to https://github.com/go-sql-driver/mysql#tls",
		tlsConfig,
	)
	config.TLSConfig = tlsConfig

	db, err := gorm.Open(gormmysql.Open(config.FormatDSN()))
	require.NoErrorf(cli.su.T(), err, "open db error: %s", err)

	return db
}

func (cli *Cli) ConnectDB(orgID, prjID, clusterID, user, password, database, tlsConfig string) *gorm.DB {
	cluster := cli.GetCluster(orgID, prjID, clusterID)

	config := mysql.NewConfig()
	config.Net = "tcp"
	config.User = user
	config.Addr = fmt.Sprintf("%s:%d", *cluster.Endpoint.Address, *cluster.Endpoint.Port)
	config.Passwd = password
	config.DBName = database
	config.ParseTime = true
	config.Timeout = 30 * time.Second
	config.ReadTimeout = 30 * time.Second
	config.WriteTimeout = 30 * time.Second
	config.Params = map[string]string{"charset": "utf8mb4"}
	config.Collation = "utf8mb4_bin"

	// Only allow TLS config key to be one of the following values.
	// Currently, not support customized server CA or enable mTLS (server verify client).
	require.Containsf(
		cli.su.T(),
		[]string{"", "true", "false", "skip-verify", "preferred"},
		tlsConfig,
		"tls mod not supported: %s, refer to https://github.com/go-sql-driver/mysql#tls",
		tlsConfig,
	)
	config.TLSConfig = tlsConfig

	db, err := gorm.Open(gormmysql.Open(config.FormatDSN()))
	require.NoErrorf(cli.su.T(), err, "open db error: %s", err)

	return db
}

func (cli *Cli) ExecuteSql(orgID, prjID, clusterID, userPrefix, password, tlsConfig string) error {
	cluster := cli.GetCluster(orgID, prjID, clusterID)

	return sql.ExecuteSqlWithUrl(
		context.Background(),
		*cluster.Endpoint.Address,
		userPrefix,
		password,
		tlsConfig,
		*cluster.Endpoint.Port,
	)
}

func (cli *Cli) DeleteClusterAllowBizErr(orgID, prjID, clusterID string) *model.CentralDefaultResp {
	params := cluster_service.NewDeleteClusterParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjID)
	params.SetClusterID(clusterID)

	resp, err := cli.ClusterService.DeleteCluster(params, nil)
	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))
	return resp.Payload
}

func (cli *Cli) EnsureDeleteCluster(orgID, prjID, clusterName string) {
	clusters := cli.ListClusters(orgID, prjID, 1, 100)

	cluster, ok := lo.Find(clusters, func(c *model.CentralCluster) bool {
		return c != nil && c.ID != nil && c.Name != nil && *c.Name == clusterName
	})

	if !ok {
		// cluster not found, nothing to do
		return
	}
	resp := cli.DeleteClusterAllowBizErr(orgID, prjID, *cluster.ID)
	require.Falsef(
		cli.su.T(),
		resp.BaseResp != nil && resp.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)
}

func (cli *Cli) CreatePrivateLinkService(orgID, prjectID, clusterID string) {
	params := cluster_service.NewCreatePrivateLinkServiceParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjectID)
	params.SetClusterID(clusterID)

	resp, err := cli.ClusterService.CreatePrivateLinkService(params, nil)

	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))
	require.Falsef(
		cli.su.T(),
		resp.Payload.BaseResp != nil && resp.Payload.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)
}

func (cli *Cli) GetPrivateLinkService(orgID, prjectID, clusterID string) *model.CentralPrivateLinkService {
	params := cluster_service.NewGetPrivateLinkServiceParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjectID)
	params.SetClusterID(clusterID)

	resp, err := cli.ClusterService.GetPrivateLinkService(params, nil)

	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))
	require.Falsef(
		cli.su.T(),
		resp.Payload.BaseResp != nil && resp.Payload.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)

	return resp.Payload.PrivatelinkService
}

func (cli *Cli) CreatePrivateLinkEndpoint(
	orgID, projectID, clusterID, endpointName string,
) *model.CentralPrivateLinkEndpoint {
	params := cluster_service.NewCreatePrivateLinkEndpointParams()
	params.SetOrgID(orgID)
	params.SetProjectID(projectID)
	params.SetClusterID(clusterID)
	params.SetPrivatelinkEndpoint(&model.CentralPrivateLinkEndpoint{
		EndpointName: endpointName,
	})
	resp, err := cli.ClusterService.CreatePrivateLinkEndpoint(params, nil)

	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))
	require.Falsef(
		cli.su.T(),
		resp.Payload.BaseResp != nil && resp.Payload.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)

	return resp.Payload.PrivatelinkEndpoint
}

func (cli *Cli) DeletePrivateLinkEndpoint(orgID, projectID, clusterID, endpointID string) {
	params := cluster_service.NewDeletePrivateLinkEndpointParams()
	params.SetOrgID(orgID)
	params.SetProjectID(projectID)
	params.SetClusterID(clusterID)
	params.SetEndpointID(endpointID)
	resp, err := cli.ClusterService.DeletePrivateLinkEndpoint(params, nil)

	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))
	require.Falsef(
		cli.su.T(),
		resp.Payload.BaseResp != nil && resp.Payload.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)
}

func (cli *Cli) CreateAWSVpcPeering(
	orgID, projectID, region, peerRegion, tenantAccountID, tenantVpcCidr, tenantVpcID string,
) *model.CentralCreateAWSVpcPeeringResp {
	params := cluster_service.NewCreateAWSVpcPeeringParams()
	params.SetOrgID(orgID)
	params.SetProjectID(projectID)
	params.SetBody(cluster_service.CreateAWSVpcPeeringBody{
		Region:          lo.ToPtr(region),
		TenantAccountID: lo.ToPtr(tenantAccountID),
		TenantRegion:    lo.ToPtr(peerRegion),
		TenantVpcCidr:   lo.ToPtr(tenantVpcCidr),
		TenantVpcID:     lo.ToPtr(tenantVpcID),
	})
	resp, err := cli.ClusterService.CreateAWSVpcPeering(params, nil)

	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))
	require.Falsef(
		cli.su.T(),
		resp.Payload.BaseResp != nil && resp.Payload.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)

	return resp.Payload
}

func (cli *Cli) GetAwsVpcPeering(orgID, projectID, vpcPeeringID string) *model.CentralAwsVpcPeering {
	params := cluster_service.NewGetAwsVpcPeeringParams()
	params.SetOrgID(orgID)
	params.SetProjectID(projectID)
	params.SetID(vpcPeeringID)

	resp, err := cli.ClusterService.GetAwsVpcPeering(params, nil)
	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))

	return resp.Payload
}

func (cli *Cli) DeleteAwsVpcPeering(orgID, projectID, vpcPeeringID string) {
	params := cluster_service.NewDeleteAwsVpcPeeringParams()
	params.SetOrgID(orgID)
	params.SetProjectID(projectID)
	params.SetID(vpcPeeringID)

	resp, err := cli.ClusterService.DeleteAwsVpcPeering(params, nil)
	require.NoError(cli.su.T(), err)
	require.True(cli.su.T(), resp.IsSuccess())
}

func (cli *Cli) CreateGcpVpcPeering(
	orgID, projectID, region, peerRegion, tenantProjectID, tenantVpcCidr, tenantNetworkName string,
) *model.CentralCreateGcpVpcPeeringResp {
	params := cluster_service.NewCreateGcpVpcPeeringParams()
	params.SetOrgID(orgID)
	params.SetProjectID(projectID)
	params.SetBody(cluster_service.CreateGcpVpcPeeringBody{
		Region:            lo.ToPtr(region),
		TenantNetworkName: lo.ToPtr(tenantNetworkName),
		TenantProjectID:   lo.ToPtr(tenantProjectID),
		TenantVpcCidr:     lo.ToPtr(tenantVpcCidr),
		TenantVpcRegion:   lo.ToPtr(peerRegion),
	})
	resp, err := cli.ClusterService.CreateGcpVpcPeering(params, nil)

	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))
	require.Falsef(
		cli.su.T(),
		resp.Payload.BaseResp != nil && resp.Payload.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)

	return resp.Payload
}

func (cli *Cli) GetGcpVpcPeering(orgID, projectID, vpcPeeringID string) *model.CentralGcpVpcPeering {
	params := cluster_service.NewGetGcpVpcPeeringParams()
	params.SetOrgID(orgID)
	params.SetProjectID(projectID)
	params.SetID(vpcPeeringID)

	resp, err := cli.ClusterService.GetGcpVpcPeering(params, nil)
	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))

	return resp.Payload
}

func (cli *Cli) ListAwsVpcPeerings(orgID, projectID string) *model.CentralAwsVpcPeerings {
	params := cluster_service.NewListAwsVpcPeeringsParams()
	params.SetOrgID(orgID)
	params.SetProjectID(projectID)

	resp, err := cli.ClusterService.ListAwsVpcPeerings(params, nil)
	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))

	return resp.Payload
}

func (cli *Cli) ListGcpVpcPeerings(orgID, projectID string) *model.CentralGcpVpcPeerings {
	params := cluster_service.NewListGcpVpcPeeringsParams()
	params.SetOrgID(orgID)
	params.SetProjectID(projectID)

	resp, err := cli.ClusterService.ListGcpVpcPeerings(params, nil)
	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))

	return resp.Payload
}

func (cli *Cli) DeleteGcpVpcPeering(orgID, projectID, vpcPeeringID string) {
	params := cluster_service.NewDeleteGcpVpcPeeringParams()
	params.SetOrgID(orgID)
	params.SetProjectID(projectID)
	params.SetID(vpcPeeringID)

	resp, err := cli.ClusterService.DeleteGcpVpcPeering(params, nil)
	require.NoError(cli.su.T(), err)
	require.True(cli.su.T(), resp.IsSuccess())
}

func (cli *Cli) PauseCluster(orgID, prjID, clusterID string) {
	params := cluster_service.NewPauseClusterParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjID)
	params.SetClusterID(clusterID)

	resp, err := cli.ClusterService.PauseCluster(params, nil)
	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))
	require.Falsef(
		cli.su.T(),
		resp.Payload.BaseResp != nil && resp.Payload.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)
}

func (cli *Cli) ResumeCluster(orgID, prjID, clusterID string) {
	params := cluster_service.NewResumeClusterParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjID)
	params.SetClusterID(clusterID)

	resp, err := cli.ClusterService.ResumeCluster(params, nil)
	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))
	require.Falsef(
		cli.su.T(),
		resp.Payload.BaseResp != nil && resp.Payload.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)
}

func (cli *Cli) GetTiDBNodes(orgID, prjectID, clusterID string) *model.CentralTiDBNodes {
	params := cluster_service.NewGetTiDBNodesParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjectID)
	params.SetClusterID(clusterID)

	resp, err := cli.ClusterService.GetTiDBNodes(params, nil)
	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))

	return resp.Payload
}

func (cli *Cli) GetTiFlashNodes(orgID, prjectID, clusterID string) *model.CentralTiFlashNodes {
	params := cluster_service.NewGetTiFlashNodesParams()
	params.SetOrgID(orgID)
	params.SetProjectID(prjectID)
	params.SetClusterID(clusterID)

	resp, err := cli.ClusterService.GetTiFlashNodes(params, nil)
	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))

	return resp.Payload
}

func (cli *Cli) CreateAwsCmek(orgID, projectID, region, kmsArn string) {
	params := cluster_service.NewCreateAwsCmekParams()
	params.SetOrgID(orgID)
	params.SetProjectID(projectID)
	params.SetBody(cluster_service.CreateAwsCmekBody{
		Specs: []*model.CentralAwsCmekSpec{
			{
				KmsArn:     kmsArn,
				RegionName: region,
			},
		},
	})

	resp, err := cli.ClusterService.CreateAwsCmek(params, nil)
	require.NoError(cli.su.T(), err)
	require.NotEmptyf(cli.su.T(), resp.Payload, "invalid response: %s", cli.RespJson(resp))
	require.Falsef(
		cli.su.T(),
		resp.Payload.BaseResp != nil && resp.Payload.BaseResp.ErrCode != "0",
		"invalid response: %s",
		cli.RespJson(resp),
	)
}