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"
)

// DeregisterImportClusterJob deregister clusterJob on import task.
func DeregisterImportClusterJob(ctx context.Context, clusterID, importID uint64) bizerr.BizError {
	return deregisterClusterJob(ctx, clusterID, importID, core.ClusterJobType_CLUSTER_JOB_TYPE_IMPORT)
}

func deregisterClusterJob(ctx context.Context, clusterID, taskID uint64, jobType core.ClusterJobType) bizerr.BizError {
	// Notes: ClusterJob has two unique keys to locate,
	// one is JobId and the other is ClusterId + JobType + Name.
	// When logging out, choose the latter
	deleteClusterJobReq := &core.DeleteClusterJobReq{
		Identifier: &core.ClusterJobIdentifier{
			JobId:     0,
			ClusterId: clusterID,
			JobType:   jobType,
			Name:      strconv.FormatUint(taskID, constants.DecimalBase),
		},
	}

	resp, grpcErr := grpcli.DeleteClusterJob(ctx, deleteClusterJobReq)
	if grpcErr != nil {
		log.L().Error("failed to deregister clusterJob, grpcErr",
			zap.Any("deleteClusterJobReq", deleteClusterJobReq),
			zap.Error(grpcErr))

		return grpcErr
	}
	if resp.BaseResp != nil && resp.BaseResp.ErrCode != 0 {
		log.L().Error("failed to deregister clusterJob, respErr",
			zap.Uint64("clusterID", clusterID),
			zap.Any("deleteClusterJobReq", deleteClusterJobReq),
			zap.Any("deleteClusterJobResp", resp),
		)
		return bizerr.New(bizerr.ErrCode(resp.BaseResp.ErrCode), "%s", resp.BaseResp.ErrMsg)
	}

	return nil
}

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

func deregisterPauseLock(
	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")
	}

	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 err
	}

	if !ss.IsExisted {
		return nil
	}

	for _, status := range ss.ExistedCustomStatuses {
		if status.StatusName != statusName {
			continue
		}
		err = mgmtSvcCli.DeleteClusterCustomStatus(ctx, &mgmt.DeleteClusterCustomStatusRequest{
			ClusterId: clusterID,
			StatusId:  status.StatusId,
		})
		if err != nil {
			log.L().Error("failed to deregister pause lock",
				zap.Uint64("clusterID", clusterID),
				zap.String("statusName", statusName),
				zap.Uint64("statusID", status.StatusId),
				zap.String("statusType", statusType.String()),
				zap.Error(err))
			return err
		}
	}

	return nil
}