//go:generate ${GOBIN}/mockgen -destination client_mock.go -package=serverlesscli github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc RegionalK8SOperationServiceClient,ClusterCROperationServiceClient,ChangefeedOpsServiceClient
package serverlesscli

import (
	"context"
	"fmt"
	"math"

	"github.com/go-logr/logr"
	"github.com/tidbcloud/devops/pkg/infra/utils/grpcutil"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/serverless"
	"google.golang.org/grpc"
)

type Client struct {
	billingCli            serverless.ServerlessBillingServiceClient
	branchCli             rpc.BranchServiceClient
	clusterCROperationCli rpc.ClusterCROperationServiceClient
	regionalK8SOpsCli     rpc.RegionalK8SOperationServiceClient
	changefeedOpsCli      rpc.ChangefeedOpsServiceClient
	logger                logr.Logger
}

// NewTestClient creates a Client with injected gRPC clients for testing.
func NewTestClient(regional rpc.RegionalK8SOperationServiceClient, crOps rpc.ClusterCROperationServiceClient, changefeedOps ...rpc.ChangefeedOpsServiceClient) *Client {
	client := &Client{regionalK8SOpsCli: regional, clusterCROperationCli: crOps}
	if len(changefeedOps) > 0 {
		client.changefeedOpsCli = changefeedOps[0]
	}
	return client
}

func NewClient(addr string, logger logr.Logger) (*Client, error) {
	c := &Client{
		logger: logger.WithName("ServerlessCli"),
	}

	opts := grpcutil.PresetDailOption().Default()
	serverlessConn, err := grpc.NewClient(addr, opts...)
	if err != nil {
		return nil, fmt.Errorf("dial serverless-service %s failed: %s", addr, err)
	}
	c.billingCli = serverless.NewServerlessBillingServiceClient(serverlessConn)
	c.branchCli = rpc.NewBranchServiceClient(serverlessConn)
	c.clusterCROperationCli = rpc.NewClusterCROperationServiceClient(serverlessConn)
	c.regionalK8SOpsCli = rpc.NewRegionalK8SOperationServiceClient(serverlessConn)
	c.changefeedOpsCli = rpc.NewChangefeedOpsServiceClient(serverlessConn)

	return c, nil
}

func (c *Client) GetSpendLimitDetails(ctx context.Context, orgID, clusterID uint64) (*serverless.GetSpendLimitDetailsResp, error) {
	resp, err := c.billingCli.GetSpendLimitDetails(ctx, &serverless.GetSpendLimitDetailsReq{
		OrgId:     orgID,
		ClusterId: clusterID,
	})
	if err != nil {
		return nil, err
	}
	if err := grpcutil.RespToErrServerless(resp.BaseResp); err != nil {
		return nil, err
	}
	return resp, nil
}

func (c *Client) ListBranches(ctx context.Context, clusterID uint64) (*rpc.ListBranchesResp, error) {
	resp, err := c.branchCli.ListBranches(ctx, &rpc.ListBranchesReq{
		ClusterId: clusterID,
	})
	if err != nil {
		return nil, err
	}
	if err := grpcutil.RespToErrServerless(resp.BaseResp); err != nil {
		return nil, err
	}
	return resp, nil
}

func (c *Client) UpdateCluster(ctx context.Context, request *rpc.UpdateClusterRequest) (*rpc.ClusterDetails, error) {
	cluster, err := c.clusterCROperationCli.UpdateCluster(ctx, request)
	if err != nil {
		return nil, err
	}
	return cluster, nil
}

func (c *Client) GetCluster(ctx context.Context, clusterID, projectID, orgID string) (*rpc.ClusterDetails, error) {
	request := &rpc.GetClusterRequest{
		ClusterId: clusterID,
		ProjectId: projectID,
		OrgId:     orgID,
	}
	cluster, err := c.clusterCROperationCli.GetCluster(ctx, request)
	if err != nil {
		return nil, err
	}
	return cluster, nil
}

// ExecuteCommand forwards a command (e.g. pd-ctl) to the ops-service on the
// specified k8s cluster via serverless-service's RegionalK8SOperationService.
func (c *Client) ExecuteCommand(ctx context.Context, provider, region, k8sCluster, command string) (*rpc.OpsExecuteCommandResponse, error) {
	resp, err := c.regionalK8SOpsCli.ExecuteCommand(ctx, &rpc.OpsExecuteCommandRequest{
		Provider:   provider,
		Region:     region,
		K8SCluster: k8sCluster,
		Command:    command,
	})
	if err != nil {
		return nil, err
	}
	return resp, nil
}

// RestartTiDBPodsByNamespace restarts TiDB pods in the given namespace via
// serverless-service's RegionalK8SOperationService. force=true asks ops-service
// to delete pods with GracePeriodSeconds=1 (skip graceful drain).
func (c *Client) RestartTiDBPodsByNamespace(ctx context.Context, provider, region, k8sCluster, namespace string, force bool) (*rpc.OpsRestartByNamespaceResponse, error) {
	resp, err := c.regionalK8SOpsCli.RestartTiDBPodsByNamespace(ctx, &rpc.OpsRestartByNamespaceRequest{
		Provider:   provider,
		Region:     region,
		K8SCluster: k8sCluster,
		Namespace:  namespace,
		Force:      force,
	})
	if err != nil {
		return nil, err
	}
	return resp, nil
}

// RestartTiDBPodsByCluster restarts TiDB pods assigned to the given keyspace
// via serverless-service's RegionalK8SOperationService. force=true asks
// ops-service to delete pods with GracePeriodSeconds=1 (skip graceful drain).
func (c *Client) RestartTiDBPodsByCluster(ctx context.Context, provider, region, k8sCluster, keyspaceToken string, force bool) (*rpc.OpsRestartByClusterResponse, error) {
	resp, err := c.regionalK8SOpsCli.RestartTiDBPodsByCluster(ctx, &rpc.OpsRestartByClusterRequest{
		Provider:      provider,
		Region:        region,
		K8SCluster:    k8sCluster,
		KeyspaceToken: keyspaceToken,
		Force:         force,
	})
	if err != nil {
		return nil, err
	}
	return resp, nil
}

func (c *Client) ListClusterChangefeeds(ctx context.Context, clusterID string, limit, offset uint32) (*rpc.ListClusterChangefeedsResponse, error) {
	if c.changefeedOpsCli == nil {
		return nil, fmt.Errorf("changefeed ops client is not configured")
	}
	if limit > math.MaxInt32 {
		return nil, fmt.Errorf("limit %d exceeds maximum supported value %d", limit, math.MaxInt32)
	}
	return c.changefeedOpsCli.ListClusterChangefeeds(ctx, &rpc.ListClusterChangefeedsRequest{
		ClusterId: clusterID,
		Limit:     int32(limit),
		Offset:    int64(offset),
	})
}

func (c *Client) ApplyChangefeedConfigChanges(ctx context.Context, req *rpc.ApplyConfigChangesRequest) (*rpc.ApplyConfigChangesResponse, error) {
	if c.changefeedOpsCli == nil {
		return nil, fmt.Errorf("changefeed ops client is not configured")
	}
	return c.changefeedOpsCli.ApplyConfigChanges(ctx, req)
}

func (c *Client) GetChangefeed(ctx context.Context, changefeedID string) (*rpc.ChangefeedInfo, error) {
	if c.changefeedOpsCli == nil {
		return nil, fmt.Errorf("changefeed ops client is not configured")
	}
	return c.changefeedOpsCli.GetChangefeed(ctx, &rpc.GetChangefeedRequest{
		ClusterId:    changefeedID,
		ChangefeedId: changefeedID,
	})
}

func (c *Client) PauseChangefeed(ctx context.Context, changefeedID string) (*rpc.ChangefeedInfo, error) {
	if c.changefeedOpsCli == nil {
		return nil, fmt.Errorf("changefeed ops client is not configured")
	}
	return c.changefeedOpsCli.PauseChangefeed(ctx, &rpc.PauseChangefeedRequest{
		ClusterId:    changefeedID,
		ChangefeedId: changefeedID,
	})
}

func (c *Client) ResumeChangefeed(ctx context.Context, changefeedID string) (*rpc.ChangefeedInfo, error) {
	if c.changefeedOpsCli == nil {
		return nil, fmt.Errorf("changefeed ops client is not configured")
	}
	return c.changefeedOpsCli.ResumeChangefeed(ctx, &rpc.ResumeChangefeedRequest{
		ClusterId:    changefeedID,
		ChangefeedId: changefeedID,
	})
}

func (c *Client) RestartChangefeed(ctx context.Context, changefeedID string) (*rpc.ChangefeedInfo, error) {
	if c.changefeedOpsCli == nil {
		return nil, fmt.Errorf("changefeed ops client is not configured")
	}
	return c.changefeedOpsCli.RestartChangefeed(ctx, &rpc.RestartChangefeedRequest{
		ClusterId:    changefeedID,
		ChangefeedId: changefeedID,
	})
}