package cluster_job

import (
	"context"
	"strconv"

	"github.com/pingcap/log"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/internal/constants"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

// GetImportClusterJob get clusterJob on import task.
func GetImportClusterJob(ctx context.Context, clusterID, importID uint64) (*core.ClusterJob, bizerr.BizError) {
	return getClusterJob(ctx, clusterID, importID, core.ClusterJobType_CLUSTER_JOB_TYPE_IMPORT)
}

func getClusterJob(ctx context.Context, clusterID, taskID uint64, jobType core.ClusterJobType) (*core.ClusterJob, bizerr.BizError) {
	listClusterJobReq := &core.ListClusterJobsReq{
		Filter: &core.ClusterJobFilter{
			JobType:    jobType,
			ClusterIds: []uint64{clusterID},
		},
	}

	resp, grpcErr := grpcli.ListClusterJobs(ctx, listClusterJobReq)
	if grpcErr != nil {
		log.L().Error("failed to list clusterJob, grpcErr",
			zap.Any("listClusterJobReq", listClusterJobReq),
			zap.Error(grpcErr))

		return nil, grpcErr
	}
	if resp.BaseResp != nil && resp.BaseResp.ErrCode != 0 {
		log.L().Error("failed to list clusterJob, respErr",
			zap.Uint64("clusterID", clusterID),
			zap.Any("listClusterJobReq", listClusterJobReq),
			zap.Any("listClusterJobResp", resp),
		)
		return nil, bizerr.New(bizerr.ErrCode(resp.BaseResp.ErrCode), "%s", resp.BaseResp.ErrMsg)
	}
	taskName := strconv.FormatUint(taskID, constants.DecimalBase)
	for _, job := range resp.GetJobs() {
		if job.GetName() == taskName {
			return job, nil
		}
	}
	return nil, nil
}