package impl

import (
	"context"
	"strconv"

	"github.com/AlekSi/pointer"
	workflowEntity "github.com/shiningrush/fastflow/pkg/entity"
	appconfig "github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/internal/feature_gate"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	restore_model "github.com/tidbcloud/dataflow-service/app/models/restore"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	restore_repo "github.com/tidbcloud/dataflow-service/app/repo/restore"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	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"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
)

type snapshotRestore struct {
	baseRestoreModel

	backupRepo backup.BackupRepo

	workflowClient workflow.Workflow
}

func NewSnapshotRestore(
	dbProvider db.TransProvider,
	backupRepo backup.BackupRepo,
	restoreRepo restore_repo.RestoreRepo,
	bucketRepo repo.BucketRepo,
	coreCli grpcli.CoreGrpcClient,
	mgmtSvcCli grpcli.MgmtSvcGrpcClient,
	dedicatedGlobalClient grpcli.DedicatedGlobalGrpcClient,
	configCenterCli grpcli.ConfigCenterGrpcClient,
	workflowClient workflow.Workflow,
	featureGate feature_gate.FeatureGate,
) restore_model.RestoreModel {
	s := &snapshotRestore{
		baseRestoreModel: baseRestoreModel{
			dbProvider:            dbProvider,
			restoreRepo:           restoreRepo,
			bucketRepo:            bucketRepo,
			coreCli:               coreCli,
			mgmtSvcCli:            mgmtSvcCli,
			dedicatedGlobalClient: dedicatedGlobalClient,
			configCenterCli:       configCenterCli,
			featureGate:           featureGate,
		},
		backupRepo:     backupRepo,
		workflowClient: workflowClient,
	}
	s.statusMatrix = NewRestoreStatusMatrix()
	s.createRestoreHook = s
	s.deleteRestoreHook = s
	s.baseRestoreModel.createRestoreParamBuilders = []createRestoreParamBuilder{
		s.snapshotBackupParamBuilder,
		s.bucketParamBuilder,
		s.restoreClusterVersionBuilder,
		s.restoreClusterParamBuilder,
	}
	s.baseRestoreModel.clusterSpecValidators = []clusterSpecValidator{
		s.ComponentTikvChecker,
		s.ComponentTiFlashChecker,
	}
	return s
}

//nolint:nestif
func (r *snapshotRestore) makeCreateRestoreEntity(
	ctx context.Context, createParam *createRestoreParam,
) (*entity.Restore, bizerr.BizError) {
	var (
		region                  string
		createClusterReqV2Bytes []byte
		err                     error
	)
	if createParam.createClusterReqV2 != nil {
		region = createParam.createClusterReqV2.Region
		createClusterReqV2Bytes, err = protojson.Marshal(createParam.createClusterReqV2)
		if err != nil {
			log.GetLogger(ctx).Error("failed to marshal create cluster request v2", zap.Error(err))
			return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
		}
	} else if createParam.createRestoreReq.ClusterConfig.ClusterID != 0 {
		region = createParam.createRestoreReq.ClusterConfig.Region
	} else {
		return nil, bizerr_wrapper.NewInternalErr(
			"createClusterReq or createClusterReqV2 is required when cluster id is not specified")
	}

	req := createParam.createRestoreReq
	return &entity.Restore{
		Name:     req.ClusterConfig.Name,
		UserID:   entity.ID(req.UserID),
		Mode:     entity.RestoreModeSnapshot,
		Region:   &region,
		BackupID: entity.ID(req.BackupID),
		Status:   entity.RestoreStatusRUNNING,
		// RequestArgs:   string(createClusterReqBytes),
		RequestArgsV2: string(createClusterReqV2Bytes),
		KernelMetrics: &entity.RestoreKernelMetrics{
			Status: pointer.To(entity.RestoreKernelStatusPending),
		},
		TenantID:  entity.ID(req.TenantID),
		ProjectID: entity.ID(req.ProjectID),
		BucketID:  createParam.bucketID,
		ClusterID: entity.ID(req.ClusterConfig.ClusterID),
		ExtraInfo: &entity.RestoreExtra{
			RestoreCRNameVersion: "v1",
		},
	}, nil
}

func (r *snapshotRestore) runCreateRestoreDagIns(
	ctx context.Context, createParam *createRestoreParam, restore *entity.Restore,
) (*workflowEntity.DagInstance, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	// start wfs
	dagVars, err := r.buildRestoreSnapshotBackupDagVars(ctx, restore, createParam)
	if err != nil {
		logger.Error("failed to build dag vars", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	dagIns, dagErr := r.workflowClient.RunDagWithTags(
		workflow.DagSnapshotRestore, dagVars,
		map[string]string{
			workflow.BusinessTypeKey:   string(workflow.BusinessTypeRestore),
			workflow.BusinessActionKey: string(workflow.BusinessActionRestoreSnapshot),
			workflow.BusinessID:        strconv.FormatUint(uint64(restore.ID), 10),
			workflow.ClusterID:         strconv.FormatUint(uint64(createParam.backup.ClusterID), 10),
		})
	if dagErr != nil {
		logger.Error("failed to run dag ins", zap.Error(dagErr))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	logger.Info("create snapshot restore dag started",
		zap.Any("restore id", restore.ID), zap.Any("dag id", dagIns.ID))
	return dagIns, nil
}

func (r *snapshotRestore) restoreClusterVersionBuilder(
	ctx context.Context, req *createRestoreParam,
) bizerr.BizError {
	req.createClusterVersion = req.backup.ClusterVersion
	return nil
}

func (r *snapshotRestore) snapshotBackupParamBuilder(
	ctx context.Context, param *createRestoreParam,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	req := param.createRestoreReq
	backup, err := r.backupRepo.Load(ctx, req.BackupID)
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			logger.Warn("backup not found", zap.Uint64("backup_id", req.BackupID))
			return bizerr_wrapper.NewNotFoundErr(appconfig.ErrMsgBackupNotFound, req.BackupID)
		}
		logger.Error("failed to load backup", zap.Error(err))
		return bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	err = r.checkBackupInfoForCreateRestore(ctx, backup)
	if err != nil {
		return err
	}
	param.backup = backup
	return nil
}

func (r *snapshotRestore) buildRestoreSnapshotBackupDagVars(
	ctx context.Context, restore *entity.Restore, createRestoreParam *createRestoreParam,
) (map[string]string, bizerr.BizError) {
	runVar := map[string]string{}
	runVar[workflow.KeyRestoreID] = strconv.FormatUint(uint64(restore.ID), 10)
	if createRestoreParam.createRestoreReq.ClusterConfig.ClusterID == 0 {
		runVar[workflow.KeyRestoreToClusterType] = string(domain.RestoreToClusterTypeNew)
	} else {
		if createRestoreParam.createRestoreReq.ClusterConfig.Cloned {
			runVar[workflow.KeyRestoreToClusterType] = string(domain.RestoreToClusterTypeClone)
		} else {
			runVar[workflow.KeyRestoreToClusterType] = string(domain.RestoreToClusterTypeExist)
		}
	}

	if createRestoreParam.createRestoreReq.BRVersion != "" {
		runVar[workflow.BRVersion] = createRestoreParam.createRestoreReq.BRVersion
	}
	if createRestoreParam.createRestoreReq.BRImageTag != "" {
		runVar[workflow.BRImageTag] = createRestoreParam.createRestoreReq.BRImageTag
	}
	if createRestoreParam.createRestoreReq.BRCustomOptions != "" {
		runVar[workflow.BRCustomOptions] = createRestoreParam.createRestoreReq.BRCustomOptions
	}
	return runVar, nil
}

func (r *snapshotRestore) checkBackupInfoForCreateRestore(
	ctx context.Context, backup *entity.Backup,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	if backup.DeletedAt != nil {
		logger.Warn("backup is deleted", zap.Any("backup_id", backup.ID))
		return bizerr_wrapper.NewNotFoundErr(appconfig.ErrMsgBackupNotFound, backup.ID)
	}
	if backup.Status != entity.BackupStatusSuccess {
		logger.Warn("backup is not success", zap.Any("backup_id", backup.ID))
		return bizerr_wrapper.NewInvalidInputError(appconfig.ErrMsgBackupStatusIsNotAllowRestore, backup.ID)
	}
	return nil
}

func (r *snapshotRestore) buildRecycleRestoreSnapshotBackupDagVars(
	ctx context.Context, restore *entity.Restore,
) (map[string]string, bizerr.BizError) {
	runVar := map[string]string{}
	runVar[workflow.KeyRestoreID] = strconv.FormatUint(uint64(restore.ID), 10)
	return runVar, nil
}

func (r *snapshotRestore) runDeleteRestoreDagIns(
	ctx context.Context, restore *entity.Restore,
) (*workflowEntity.DagInstance, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	// start wfs
	dagVars, err := r.buildRecycleRestoreSnapshotBackupDagVars(ctx, restore)
	if err != nil {
		logger.Error("failed to build dag vars", zap.Error(err))
		return nil, err
	}
	dagIns, dagErr := r.workflowClient.RunDagWithTags(
		workflow.DagGcSnapshotRestore, dagVars,
		map[string]string{
			workflow.BusinessID:        strconv.FormatUint(uint64(restore.ID), 10),
			workflow.BusinessActionKey: string(workflow.BusinessActionGcRestoreSnapshot),
			workflow.BusinessTypeKey:   string(workflow.BusinessTypeRestore),
			workflow.ClusterID:         strconv.FormatUint(uint64(restore.ClusterID), 10),
		})
	if dagErr != nil {
		logger.Error("failed to run dag ins", zap.Error(dagErr))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	logger.Info("delete snapshot restore dag ins started",
		zap.Any("restore id", restore.ID), zap.Any("dag id", dagIns.ID))
	return dagIns, nil
}