package cluster_job

import (
	"context"
	"strconv"

	"github.com/pingcap/log"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	appcontainer "github.com/tidbcloud/dataflow-service/app/container"
	"github.com/tidbcloud/dataflow-service/internal/constants"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/pkgv2/bizerr"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const (
	dmPauseLockStatusPrefix = "dm-job-"
)

// RegisterImportClusterJob register clusterJob on import task.
func RegisterImportClusterJob(ctx context.Context, clusterID, importID uint64) bizerr.BizError {
	return registerClusterJob(ctx, clusterID, importID, core.ClusterJobType_CLUSTER_JOB_TYPE_IMPORT)
}

func registerClusterJob(ctx context.Context, clusterID, taskID uint64, jobType core.ClusterJobType) bizerr.BizError {
	createClusterJobReq := &core.CreateClusterJobReq{
		ClusterId: clusterID,
		JobType:   jobType,
		Name:      strconv.FormatUint(taskID, constants.DecimalBase),

		BaseReq: nil,
	}

	resp, grpcErr := grpcli.CreateClusterJob(ctx, createClusterJobReq)
	if grpcErr != nil {
		log.L().Error("failed to register clusterJob, grpcErr",
			zap.Any("createClusterJobReq", createClusterJobReq),
			zap.Error(grpcErr))

		return grpcErr
	}
	if resp.BaseResp != nil && resp.BaseResp.ErrCode != 0 {
		log.L().Error("failed to register clusterJob, respErr",
			zap.Uint64("clusterID", clusterID),
			zap.Any("createClusterJobReq", createClusterJobReq),
			zap.Any("createClusterJobResp", resp),
		)

		return bizerr.New(bizerr.ErrCode(resp.BaseResp.ErrCode), "%s", resp.BaseResp.ErrMsg)
	}

	return nil
}

func IsDMPauseLockRegistered(ctx context.Context, clusterID, dmJobID uint64) (bool, bizerr.BizError) {
	return isPauseLockRegistered(
		ctx,
		clusterID,
		dmPauseLockStatusName(dmJobID),
		mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_DATA_MIGRATION,
	)
}

func isPauseLockRegistered(
	ctx context.Context,
	clusterID uint64,
	statusName string,
	statusType mgmt.CustomStatusType,
) (bool, bizerr.BizError) {
	mgmtSvcCli := appcontainer.Global().MgmtSvcGrpcClient
	if mgmtSvcCli == nil {
		return false, bizerr_wrapper.NewInternalErr("mgmt svc grpc client is nil")
	}

	queryReq := &mgmt.IsClusterCustomStatusExistedRequest{
		ClusterId:  clusterID,
		StatusType: statusType,
	}
	ss, err := mgmtSvcCli.IsClusterCustomStatusExisted(ctx, queryReq)
	if err != nil {
		log.L().Error("failed to query pause lock status",
			zap.Uint64("clusterID", clusterID),
			zap.String("statusName", statusName),
			zap.String("statusType", statusType.String()),
			zap.Error(err))
		return false, err
	}

	if !ss.IsExisted {
		return false, nil
	}

	for _, status := range ss.ExistedCustomStatuses {
		if status.StatusName == statusName {
			return true, nil
		}
	}

	return false, nil
}

func RegisterDMPauseLock(ctx context.Context, clusterID, dmJobID uint64) bizerr.BizError {
	return registerPauseLock(
		ctx,
		clusterID,
		dmPauseLockStatusName(dmJobID),
		mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_DATA_MIGRATION,
	)
}

func registerPauseLock(
	ctx context.Context,
	clusterID uint64,
	statusName string,
	statusType mgmt.CustomStatusType,
) bizerr.BizError {
	mgmtSvcCli := appcontainer.Global().MgmtSvcGrpcClient
	if mgmtSvcCli == nil {
		return bizerr_wrapper.NewInternalErr("mgmt svc grpc client is nil")
	}

	setStatusReq := &mgmt.SetClusterCustomStatusRequest{
		CustomStatus: &mgmt.ClusterCustomStatus{
			ClusterId:  clusterID,
			StatusName: statusName,
			StatusType: statusType,
			Mutex: &mgmt.ClusterCustomStatus_CustomMutex{
				StatusType: []mgmt.CustomStatusType{},
				ClusterAction: []mgmt.ClusterAction{
					mgmt.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER,
				},
			},
		},
	}

	_, err := mgmtSvcCli.SetClusterCustomStatus(ctx, setStatusReq)
	if err != nil {
		log.L().Error("failed to register pause lock",
			zap.Uint64("clusterID", clusterID),
			zap.String("statusName", statusName),
			zap.String("statusType", statusType.String()),
			zap.Error(err))
		return err
	}

	return nil
}

func dmPauseLockStatusName(dmJobID uint64) string {
	return dmPauseLockStatusPrefix + strconv.FormatUint(dmJobID, constants.DecimalBase)
}