package bizutils

import (
	"context"

	"github.com/tidbcloud/pkgv2/bizerr"

	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
)

const internalErr = "Internal error, please retry later or contact support."

// GetCluster gets the cluster information if the requested user has the right permission.
func GetCluster(ctx context.Context, orgId, projectId, clusterID uint64) (*core.Cluster, error) {
	cluster, err := grpcli.GetClusterByID(ctx, clusterID)
	if err != nil {
		if bizerr.ErrCodeRecordNotFound == err.ErrCode() {
			return nil, errcode.NewNotFoundErr(errors.Errorf("Cluster id=%d not found", clusterID))
		}
		return nil, errcode.NewInternalErr(errors.New(internalErr))
	}

	if cluster.TenantId != orgId || cluster.ProjectId != projectId {
		return nil, errcode.NewForbiddenErr(errors.New("you do not have permission to access this resource"))
	}
	return cluster, nil
}

// CheckCluster checks ClusterId and ClusterStatus.
func CheckCluster(ctx context.Context, orgID, projID, clusterID uint64) error {
	cluster, err := GetCluster(ctx, orgID, projID, clusterID)
	if err != nil {
		return err
	}

	if !ClusterAvailable(cluster) {
		return errcode.NewInvalidInputErr(errors.Errorf("Cluster id=%d not in status normal/scaling, it is in %v", clusterID, cluster.Status))
	}
	return nil
}

// CheckClusterOwner checks whether the cluster's tenantId and projectId are same as input.
func CheckClusterOwner(ctx context.Context, orgId, projectId, clusterID uint64) error {
	_, err := GetCluster(ctx, orgId, projectId, clusterID)
	return err
}

// CheckClusterId checks whether the cluster's status is NORMAL/SCALING.
func CheckClusterStatus(ctx context.Context, clusterID uint64) error {
	cluster, err := grpcli.GetClusterByID(ctx, clusterID)
	if err != nil {
		if bizerr.ErrCodeRecordNotFound == err.ErrCode() {
			return errcode.NewNotFoundErr(errors.Errorf("Cluster id=%d not found", clusterID))
		}
		return err
	}

	if cluster.Status != core.ClusterStatus_CLUSTER_STATUS_NORMAL {
		return errcode.NewInvalidInputErr(errors.Errorf("Cluster ID %d is not in status normal, please try again later.", clusterID))
	}
	return nil
}

func ClusterDeletedOrCleared(c *core.Cluster) bool {
	return c.Status == core.ClusterStatus_CLUSTER_STATUS_DELETED ||
		c.Status == core.ClusterStatus_CLUSTER_STATUS_DELETECOMPLETE ||
		c.Status == core.ClusterStatus_CLUSTER_STATUS_CLEARED
}

// ClusterAvailable checks whether the cluster is available to connect to, i.e. it's in NORMAL/SCALING status
func ClusterAvailable(c *core.Cluster) bool {
	return c.Status == core.ClusterStatus_CLUSTER_STATUS_NORMAL ||
		c.Status == core.ClusterStatus_CLUSTER_STATUS_SCALING
}