// Package cluster_job provides an implementation of the Model interface
// that handles register and deregister ClusterJobs for tasks in TiDB.
// It uses the gRPC client from the pkg/grpc/grpcli package to communicate with the core service.
package cluster_job

import (
	"context"
	"strconv"

	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	appconfig "github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/internal/constants"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const (
	// MinSnowflakeID the last 22 bits of snowflake id are machine id and machine sequence number.
	MinSnowflakeID = 1 << 22

	pitrPauseLockStatusNamePrefix = "pitr-job-"
)

type Model interface {
	RegisterCDCClusterJob(ctx context.Context, clusterID, taskID uint64) bizerr.BizError
	DeregisterCDCClusterJob(ctx context.Context, clusterID, taskID uint64) bizerr.BizError

	RegisterBackupClusterJob(ctx context.Context, clusterID, taskID uint64) bizerr.BizError
	DeregisterBackupClusterJob(ctx context.Context, clusterID, backupID uint64) bizerr.BizError
	IsBackupClusterJobRegistered(ctx context.Context, clusterID, taskID uint64) bool

	DeregisterRestoreClusterJob(ctx context.Context, clusterID uint64, restoreMode entity.RestoreMode) bizerr.BizError

	RegisterRestoreClusterV2(ctx context.Context, clusterID, taskID uint64) bizerr.BizError
	DeregisterRestoreClusterJobV2(ctx context.Context, clusterID, taskID uint64) bizerr.BizError

	RegisterPiTRPauseLock(ctx context.Context, clusterID, turn uint64) bizerr.BizError
	DeregisterPiTRPauseLock(ctx context.Context, clusterID, turn uint64) bizerr.BizError
	IsPiTRPauseLockRegistered(ctx context.Context, clusterID, turn uint64) (bool, bizerr.BizError)
}

// The clusterJobModelImpl struct provides the implementation of the Model interface.
type clusterJobModelImpl struct {
	coreGrpcClient grpcli.CoreGrpcClient
	mgmtSvcCli     grpcli.MgmtSvcGrpcClient
}

func NewClusterJobModelImpl(coreGrpcClient grpcli.CoreGrpcClient, mgmtSvcCli grpcli.MgmtSvcGrpcClient) Model {
	return &clusterJobModelImpl{
		coreGrpcClient: coreGrpcClient,
		mgmtSvcCli:     mgmtSvcCli,
	}
}

func (job *clusterJobModelImpl) IsBackupClusterJobRegistered(ctx context.Context, clusterID, taskID uint64) bool {
	req := &core.IsClusterJobRegisteredReq{
		Identifier: &core.ClusterJobIdentifier{
			ClusterId: clusterID,
			JobType:   core.ClusterJobType_CLUSTER_JOB_TYPE_BACKUP,
			Name:      strconv.FormatUint(taskID, constants.DecimalBase),
		},
	}
	resp, err := job.coreGrpcClient.IsClusterJobRegistered(ctx, req)
	if err != nil {
		log.GetLogger(ctx).Error("failed to query clusterJob, grpcErr", zap.Uint64("clusterId", clusterID),
			zap.Uint64("taskId", taskID), zap.Error(err))
		return false
	}
	return resp.IsRegistered
}

// RegisterCDCClusterJob registers a cluster job for a CDC task.
func (job *clusterJobModelImpl) RegisterCDCClusterJob(ctx context.Context, clusterID, taskID uint64) bizerr.BizError {
	return job.register(ctx, clusterID, taskID, core.ClusterJobType_CLUSTER_JOB_TYPE_CDC)
}

func (job *clusterJobModelImpl) DeregisterCDCClusterJob(ctx context.Context, clusterID, taskID uint64) bizerr.BizError {
	return job.deregister(
		ctx, clusterID, strconv.FormatUint(taskID, appconfig.DecimalBase), core.ClusterJobType_CLUSTER_JOB_TYPE_CDC)
}

// RegisterBackupClusterJob register clusterJob on backup task.
func (job *clusterJobModelImpl) RegisterBackupClusterJob(
	ctx context.Context, clusterID, taskID uint64,
) bizerr.BizError {
	return job.register(ctx, clusterID, taskID, core.ClusterJobType_CLUSTER_JOB_TYPE_BACKUP)
}

func (job *clusterJobModelImpl) register(
	ctx context.Context,
	clusterID, taskID uint64,
	jobType core.ClusterJobType,
) bizerr.BizError {
	createClusterJobReq := &core.CreateClusterJobReq{
		ClusterId: clusterID,
		JobType:   jobType,
		Name:      strconv.FormatUint(taskID, appconfig.DecimalBase),
	}

	_, err := job.coreGrpcClient.CreateClusterJob(ctx, createClusterJobReq)
	if err != nil {
		log.GetLogger(ctx).Error("failed to register clusterJob",
			zap.Uint64("clusterID", clusterID),
			zap.Any("createClusterJobReq", createClusterJobReq),
			zap.Error(err))
		return err.Chain("create cluster job")
	}
	return nil
}

func (job *clusterJobModelImpl) registerIgnoreExist(
	ctx context.Context,
	clusterID, taskID uint64,
	jobType core.ClusterJobType,
) bizerr.BizError {
	checkReq := &core.IsClusterJobRegisteredReq{
		Identifier: &core.ClusterJobIdentifier{
			ClusterId: clusterID,
			JobType:   jobType,
			Name:      strconv.FormatUint(taskID, constants.DecimalBase),
		},
	}
	checkResp, err := job.coreGrpcClient.IsClusterJobRegistered(ctx, checkReq)
	if err != nil {
		return err.Chain("check cluster job registered")
	}
	if checkResp.IsRegistered {
		log.GetLogger(ctx).Info("cluster job is already registered",
			zap.Uint64("cluster_id", clusterID),
			zap.Uint64("task_id", taskID),
			zap.String("job_type", jobType.String()))
		return nil
	}

	return job.register(ctx, clusterID, taskID, jobType)
}

func (job *clusterJobModelImpl) DeregisterRestoreClusterJob(
	ctx context.Context,
	clusterID uint64,
	restoreMode entity.RestoreMode,
) bizerr.BizError {
	var jobType core.ClusterJobType
	if restoreMode == entity.RestoreModeVolumeSnapshot {
		jobType = core.ClusterJobType_CLUSTER_JOB_TYPE_VOLUME_SNAPSHOT_RESTORE
	} else {
		jobType = core.ClusterJobType_CLUSTER_JOB_TYPE_RESTORE
	}

	listJobReq := &core.ListClusterJobsReq{
		Filter: &core.ClusterJobFilter{
			JobType:    jobType,
			ClusterIds: []uint64{clusterID},
		},
	}
	listJobResp, err := job.coreGrpcClient.ListClusterJobs(ctx, listJobReq)
	if err != nil {
		return err.Chain("list restore cluster jobs error")
	}

	if len(listJobResp.Jobs) == 0 {
		return nil
	}
	if len(listJobResp.Jobs) > 1 {
		return bizerr_wrapper.NewInternalErr(
			"the count of restore cluster jobs is more than 1, which is abnormal. cluster id: %d", clusterID)
	}
	// Notes: Restore ClusterJob unique keys is ClusterID + JobType + ClusterID.
	return job.deregister(ctx, clusterID, listJobResp.Jobs[0].Name, jobType)
}

func (job *clusterJobModelImpl) deregister(
	ctx context.Context,
	clusterID uint64,
	jobName string,
	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:      jobName,
		},
	}

	_, err := job.coreGrpcClient.DeleteClusterJob(ctx, deleteClusterJobReq)
	if err != nil {
		log.GetLogger(ctx).Error("failed to deregister clusterJob",
			zap.Uint64("clusterID", clusterID),
			zap.Any("deleteClusterJobReq", deleteClusterJobReq),
			zap.Error(err))
		return err.Chain("delete cluster job")
	}
	return nil
}

func (job *clusterJobModelImpl) DeregisterBackupClusterJob(
	ctx context.Context, clusterID, backupID uint64,
) bizerr.BizError {
	return job.deregister(
		ctx, clusterID,
		strconv.FormatUint(backupID, appconfig.DecimalBase),
		core.ClusterJobType_CLUSTER_JOB_TYPE_BACKUP)
}

func (job *clusterJobModelImpl) RegisterRestoreClusterV2(
	ctx context.Context, clusterID, taskID uint64,
) bizerr.BizError {
	return job.registerCustomStatus(
		ctx,
		clusterID,
		strconv.FormatUint(taskID, appconfig.DecimalBase),
		mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_RESTORE,
		openapicommonv1beta1.ClusterState_RESTORING.Enum(),
	)
}

func (job *clusterJobModelImpl) DeregisterRestoreClusterJobV2(
	ctx context.Context, clusterID, taskID uint64,
) bizerr.BizError {
	return job.deregisterCustomStatus(
		ctx,
		clusterID,
		strconv.FormatUint(taskID, appconfig.DecimalBase),
		mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_RESTORE,
	)
}

func (job *clusterJobModelImpl) RegisterPiTRPauseLock(ctx context.Context, clusterID, turn uint64) bizerr.BizError {
	return job.registerCustomStatus(
		ctx,
		clusterID,
		pitrPauseLockStatusName(turn),
		mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_PITR,
		nil,
	)
}

func (job *clusterJobModelImpl) DeregisterPiTRPauseLock(ctx context.Context, clusterID, turn uint64) bizerr.BizError {
	return job.deregisterCustomStatus(
		ctx,
		clusterID,
		pitrPauseLockStatusName(turn),
		mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_PITR,
	)
}

func (job *clusterJobModelImpl) IsPiTRPauseLockRegistered(ctx context.Context, clusterID, turn uint64) (bool, bizerr.BizError) {
	query := &mgmt.IsClusterCustomStatusExistedRequest{
		ClusterId:  clusterID,
		StatusType: mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_PITR,
	}
	ss, err := job.mgmtSvcCli.IsClusterCustomStatusExisted(ctx, query)
	if err != nil {
		log.GetLogger(ctx).Error("failed to query pitr pause lock status",
			zap.Uint64("clusterID", clusterID),
			zap.Any("IsClusterCustomStatusExisted", query),
			zap.Error(err))
		return false, err.Chain("query pause lock status")
	}
	if !ss.IsExisted {
		return false, nil
	}
	for _, status := range ss.ExistedCustomStatuses {
		if status.StatusName == pitrPauseLockStatusName(turn) {
			return true, nil
		}
	}
	return false, nil
}

func GetBizChangefeedTaskID(ctx context.Context, bizChangefeedID uint64) uint64 {
	if bizChangefeedID > MinSnowflakeID {
		return bizChangefeedID
	}
	log.GetLogger(ctx).Info("Mapped old/migrated bizChangefeedID to new taskID scheme",
		zap.Uint64("originalID", bizChangefeedID),
		zap.Uint64("newTaskID", bizChangefeedID+MinSnowflakeID))
	// special logic for migrated changefeed, which has a smaller ID than MinSnowflakeID
	// we can not use the bizChangefeedID as taskID, that id is already registered by cdc task
	return bizChangefeedID + MinSnowflakeID
}

// TODO(ideascf): ignore existence
func (job *clusterJobModelImpl) registerCustomStatus(
	ctx context.Context,
	clusterID uint64,
	statusName string,
	statusType mgmt.CustomStatusType,
	overrideStatus *openapicommonv1beta1.ClusterState,
) bizerr.BizError {
	setStatusReq := &mgmt.SetClusterCustomStatusRequest{
		CustomStatus: &mgmt.ClusterCustomStatus{
			ClusterId:      clusterID,
			StatusName:     statusName,
			StatusType:     statusType,
			OverrideStatus: overrideStatus,
			Mutex: &mgmt.ClusterCustomStatus_CustomMutex{
				StatusType: []mgmt.CustomStatusType{},
				ClusterAction: []mgmt.ClusterAction{
					mgmt.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER,
				},
			},
		},
	}

	_, err := job.mgmtSvcCli.SetClusterCustomStatus(ctx, setStatusReq)
	if err != nil {
		log.GetLogger(ctx).Error("failed to register clusterJob",
			zap.Uint64("clusterID", clusterID),
			zap.Any("SetClusterCustomStatus", setStatusReq),
			zap.Error(err))
		return err.Chain("create cluster job")
	}
	return nil
}

func (job *clusterJobModelImpl) deregisterCustomStatus(
	ctx context.Context,
	clusterID uint64,
	statusName string,
	statusType mgmt.CustomStatusType,
) bizerr.BizError {
	query := &mgmt.IsClusterCustomStatusExistedRequest{
		ClusterId:  clusterID,
		StatusType: statusType,
	}
	ss, err := job.mgmtSvcCli.IsClusterCustomStatusExisted(ctx, query)
	if err != nil {
		log.GetLogger(ctx).Error("failed to deregister clusterJob",
			zap.Uint64("clusterID", clusterID),
			zap.Any("IsClusterCustomStatusExisted", query),
			zap.Error(err))
		return err.Chain("deregister status")
	}
	if ss.IsExisted {
		for _, status := range ss.ExistedCustomStatuses {
			if status.StatusName != statusName {
				continue
			}
			err = job.mgmtSvcCli.DeleteClusterCustomStatus(ctx, &mgmt.DeleteClusterCustomStatusRequest{
				ClusterId: clusterID,
				StatusId:  status.StatusId,
			})
			if err != nil {
				log.GetLogger(ctx).Error("failed to deregister clusterJob",
					zap.String("statusName", statusName),
					zap.Uint64("clusterID", clusterID),
					zap.Error(err),
				)
				return err.Chain("delete status")
			}
		}
	}
	return nil
}

func pitrPauseLockStatusName(turn uint64) string {
	return pitrPauseLockStatusNamePrefix + strconv.FormatUint(turn, constants.DecimalBase)
}