package impl

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/Masterminds/semver/v3"
	"github.com/elliotchance/pie/v2"
	"github.com/gorhill/cronexpr"
	"github.com/pingcap/errors"
	dagentity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/internal/feature_gate"
	"github.com/tidbcloud/dataflow-service/app/internal/status_matrix"
	"github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/br"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	bucketRepo "github.com/tidbcloud/dataflow-service/app/repo"
	backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	restoreRepo "github.com/tidbcloud/dataflow-service/app/repo/restore"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	internal_config "github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/model/util"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/blob"
	"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"
	dedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/timestamppb"
)

var _ backup.PitBackupModel = &pitBackupModel{}

type pitBackupModel struct {
	// db
	dbProvider          db.TransProvider
	pitBackupSwitchRepo backupRepo.PitBackupSwitchRepo
	logBackupRepo       backupRepo.LogBackupRepo
	backupRepo          backupRepo.BackupRepo
	scheduleRepo        backupRepo.BackupScheduleRepo
	restoreRepo         restoreRepo.RestoreRepo
	bucketRepo          bucketRepo.BucketRepo

	// sub modules
	snapshotBackupModel backup.SnapshotBackupModel
	backupSettingModel  backup.BackupSettingModel

	// internal helpers
	workflowCli        workflow.Workflow
	coreCli            grpcli.CoreGrpcClient
	mgmtCli            grpcli.MgmtSvcGrpcClient
	switchStatusMatrix status_matrix.StatusMatrix
	featureGate        feature_gate.FeatureGate
	brCommon           *br.BRCommon

	getStorageFunc         blob.GetStorageFunc
	getCheckpointForCRPiTR func(
		ctx context.Context,
		logBackup entity.LogBackup,
		region string,
		getStorageFunc blob.GetStorageFunc,
	) (int64, bool, error)
}

func NewPitBackupModel(
	dbProvider db.TransProvider,
	pitBackupSwitchRepo backupRepo.PitBackupSwitchRepo,
	logBackupRepo backupRepo.LogBackupRepo,
	backupRepo backupRepo.BackupRepo,
	restoreRepo restoreRepo.RestoreRepo,
	bucketRepo bucketRepo.BucketRepo,
	workflowCli workflow.Workflow,
	coreCli grpcli.CoreGrpcClient,
	mgmtCli grpcli.MgmtSvcGrpcClient,
	snapshotBackupModel backup.SnapshotBackupModel,
	backupSettingModel backup.BackupSettingModel,
	scheduleRepo backupRepo.BackupScheduleRepo,
	featureGate feature_gate.FeatureGate,
) backup.PitBackupModel {
	brCommon := br.NewBRCommon(bucketRepo)
	m := &pitBackupModel{
		dbProvider:          dbProvider,
		pitBackupSwitchRepo: pitBackupSwitchRepo,
		logBackupRepo:       logBackupRepo,
		backupRepo:          backupRepo,
		scheduleRepo:        scheduleRepo,
		restoreRepo:         restoreRepo,
		bucketRepo:          bucketRepo,
		workflowCli:         workflowCli,
		coreCli:             coreCli,
		mgmtCli:             mgmtCli,
		snapshotBackupModel: snapshotBackupModel,
		backupSettingModel:  backupSettingModel,
		switchStatusMatrix:  NewPitBackupSwitchStatusMatrix(),
		featureGate:         featureGate,
		brCommon:            brCommon,

		getStorageFunc:         blob.NewStorage,
		getCheckpointForCRPiTR: brCommon.GetCheckpointForCRPiTR,
	}
	return m
}

const (
	errTip           = "PiTR has internal error, please contact the administrator."
	offTip           = "PiTR is off, you can turn it on in Backup Setting."
	scheduledTip     = "PiTR will be enabled after next backup schedule %s or a manual backup."
	backupRunningTip = "PiTR will take effect after backup %s done, please wait for a moment."
	catchingTip      = "PiTR is initializing, which may take 5 minutes."
	noBackupTip      = "You will need to create a backup for recovering." // log backup running, but no backup
)

func (b *pitBackupModel) AsyncCreate(ctx context.Context,
	req *domain.AsyncCreateBackupReq,
) (*domain.AsyncCreateBackupResp, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("pitBackupModel.AsyncCreate").With(zap.Any("cluster id", req.ClusterID))
	ctxWithLogger := log.WithLogger(ctx, logger)

	if len(req.TableFilter) != 0 {
		logger.Info("non-full snapshot backup bypasses pitr flow")
		return b.snapshotBackupModel.AsyncCreate(ctxWithLogger, req)
	}

	// get cluster first
	cluster, err := b.mgmtCli.GetAnyDedicatedClusterBasicByID(ctxWithLogger, req.ClusterID)
	if err != nil {
		log.GetLogger(ctx).Error("get cluster error", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}

	// get or create pit backup switch
	pitBackupSwitch, err := b.getPitBackupSwitch(ctxWithLogger, cluster)
	if err != nil {
		logger.Error("get pit backup switch error", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}

	if pitBackupSwitch.Status == entity.PitBackupStatusEnabled {
		// log backup is already enabled, just reuse snapshot backup
		return b.snapshotBackupModel.AsyncCreate(ctxWithLogger, req)
	} else if pitBackupSwitch.Status == entity.PitBackupStatusWaitingForEnabling {
		// found the enabling flag, we need to create a new dag to wait for the log backup enabled
		return b.enablePitBackup(ctxWithLogger, pitBackupSwitch, cluster, req)
	} else if pitBackupSwitch.Status == entity.PitBackupStatusWaitingForRestarting {
		// found the need restart flag, we need to create a new dag to wait for the log backup restarted
		return b.restartPitBackup(ctxWithLogger, pitBackupSwitch, cluster, req)
	}
	// other status should not happen
	logger.Warn("pit backup status is disabling or disabled, create snapshot backup should not use this model")
	return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
}

func (b *pitBackupModel) enablePitBackup(
	ctx context.Context,
	pitBackupSwitch *entity.PitBackupSwitch,
	cluster *mgmt.Cluster,
	req *domain.AsyncCreateBackupReq) (
	*domain.AsyncCreateBackupResp, bizerr.BizError,
) {
	if err := b.preCheckOfEnablePitBackup(ctx, cluster, pitBackupSwitch, req); err != nil {
		return nil, err
	}

	if req.DryRun {
		log.GetLogger(ctx).Info("dry run enable pit backup success", zap.Any("req", req))
		return nil, nil
	}

	resp := &domain.AsyncCreateBackupResp{}
	txF := func(ctx context.Context) error {
		// set status to doing_backup, mark this enable operation is in progress and can not be revoked
		nextStatus := b.switchStatusMatrix.CalculateNextStatusForAction(status_matrix.CurrentStatus{
			Status:  string(pitBackupSwitch.Status),
			Actions: pitBackupSwitch.Action.CurrentActions,
		}, string(entity.PitBackupSwitchActionDoEnable))
		pitBackupSwitch.Action.CurrentActions = nextStatus.Actions
		pitBackupSwitch.Status = entity.PitBackupSwitchStatus(nextStatus.Status)
		if err := b.pitBackupSwitchRepo.Save(ctx, pitBackupSwitch); err != nil {
			log.GetLogger(ctx).Error("save pit backup switch failed", zap.Error(err))
			return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
		}
		logBackup, createErr := b.createLogBackupEntity(ctx, cluster)
		if createErr != nil {
			log.GetLogger(ctx).Error("create log backup error", zap.Error(createErr))
			return createErr
		}
		if dagID, err := b.runEnablePitBackupDag(ctx, req, logBackup, pitBackupSwitch); err != nil {
			log.GetLogger(ctx).Error("run enable pit backup dag failed", zap.Error(err))
			return err
		} else {
			resp.WorkflowID = dagID
			resp.LogBackupID = logBackup.ID.Uint64()
			resp.BackupScheduleID = logBackup.BackupScheduleID.Uint64()
		}
		return nil
	}

	if err := b.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		return txF(ctx)
	}); err != nil {
		log.GetLogger(ctx).Error("enable pit backup failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	return resp, nil
}

func (b *pitBackupModel) getPitBackupSwitch(
	ctx context.Context, cluster *mgmt.Cluster,
) (*entity.PitBackupSwitch, bizerr.BizError) {
	filterF := func(qs entity.PitBackupSwitchQuerySet) entity.PitBackupSwitchQuerySet {
		qs = qs.ClusterIDEq(entity.ID(cluster.GetClusterId())).
			ProjectIDEq(entity.ID(cluster.GetProjectId())).
			TenantIDEq(entity.ID(cluster.GetOrganizationId()))
		return qs
	}
	entList, err := b.pitBackupSwitchRepo.List(ctx, filterF)
	if err != nil {
		log.GetLogger(ctx).Error("list pit backup switch", zap.Error(err))
		return nil, err
	}
	// we have the cluster unique idx, so the entList should be 0 or 1
	if len(entList) != 1 {
		log.GetLogger(ctx).Error("list pit backup switch error", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	ent := entList[0]
	return &ent, nil
}

func (b *pitBackupModel) preCheckOfEnablePitBackup(
	ctx context.Context, cluster *mgmt.Cluster,
	pitBackupSwitch *entity.PitBackupSwitch, req *domain.AsyncCreateBackupReq,
) bizerr.BizError {
	logger := log.GetLogger(ctx)

	// check can create snapshot backup
	cloneReq, cErr := req.Clone()
	if cErr != nil {
		logger.Error("clone req error", zap.Error(cErr))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	cloneReq.DryRun = true
	_, err := b.snapshotBackupModel.AsyncCreate(ctx, cloneReq)
	if err != nil {
		return err
	}

	// 0. check pit backup switch status
	isFriendly := b.switchStatusMatrix.IsActionFriendly2Status(status_matrix.CurrentStatus{
		Status:  string(pitBackupSwitch.Status),
		Actions: pitBackupSwitch.Action.CurrentActions,
	}, string(entity.PitBackupSwitchActionDoEnable))
	if !isFriendly {
		logger.Warn("do pit backup failed , status is not friendly",
			zap.Any("pit backup switch", pitBackupSwitch))
		return bizerr_wrapper.NewInvalidInputError(
			config.ErrActionParam, pitBackupSwitch.Status, entity.PitBackupSwitchActionDoEnable)
	}

	// 1. check cluster version
	// This doesn't usually happen unless user downgrade the cluster after turning on PiTR backup
	dc := cluster.GetDedicatedCluster()
	v, cErr := semver.NewVersion(dc.GetVersion())
	if cErr != nil {
		log.GetLogger(ctx).Error("parse cluster version error", zap.Error(cErr))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	backupCfg := internal_config.Global().BackupConfig
	if !backupCfg.LogBackupEnableMinVerConstraint.Check(v) {
		return bizerr.New(
			config.ErrCodePitBackupPreCheckClusterVersion,
			config.ErrMsgPitBackupPreCheckClusterVersion,
			cluster.GetClusterId(), dc.GetVersion(), backupCfg.LogBackupEnableMinVer)
	}

	// 2. check the cluster status
	if dc.GetState() != dedicatedv1.Cluster_AVAILABLE {
		return bizerr.New(
			config.ErrCodePitBackupPreCheckClusterStatus,
			config.ErrMsgPitBackupPreCheckClusterStatus,
			cluster.GetClusterId(), dc.GetState())
	}

	// 3. check the log backup status
	return b.checkActiveLogBackupExisted(ctx, cluster)
}

func (b *pitBackupModel) checkActiveLogBackupExisted(
	ctx context.Context, cluster *mgmt.Cluster,
) bizerr.BizError {
	filterF := func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
		qs = qs.ClusterIDEq(entity.ID(cluster.GetClusterId())).
			ProjectIDEq(entity.ID(cluster.GetProjectId())).
			TenantIDEq(entity.ID(cluster.GetOrganizationId())).
			DeletedAtIsNull().
			StatusNotIn(entity.LogBackupStatusStopped).
			PreloadBackupSchedule()
		return qs
	}
	logBackups, err := b.logBackupRepo.List(ctx, filterF)
	if err != nil {
		log.GetLogger(ctx).Error("list log backup error", zap.Error(err))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	if len(logBackups) == 0 {
		return nil
	}
	log.GetLogger(ctx).Error("already have active log backup(s)", zap.Any("log backups", logBackups))
	return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
}

func (b *pitBackupModel) createLogBackupEntity(
	ctx context.Context, cluster *mgmt.Cluster,
) (*entity.LogBackup, error) {
	settingResp, settingErr := b.backupSettingModel.QueryBackupSetting(ctx, pointer.ToUint64(cluster.GetClusterId()))
	if settingErr != nil {
		log.GetLogger(ctx).Error("failed to get backup setting", zap.Error(settingErr))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	logBackup := &entity.LogBackup{
		ClusterID:             entity.ID(cluster.GetClusterId()),
		ProjectID:             entity.ID(cluster.GetProjectId()),
		TenantID:              entity.ID(cluster.GetOrganizationId()),
		Status:                entity.LogBackupStatusScheduled,
		CleanPolicy:           entity.BackupCleanPolicyRetain,
		PiTRSettingSnapshot:   settingResp.BackupSetting.PiTR,
		BucketSettingSnapshot: settingResp.BackupSetting.Bucket,
		ExtraInfo: &entity.LogBackupExtra{
			On3LayerArch:        true,
			BackupCRNameVersion: "v1",
		},
	}
	if err := b.logBackupRepo.Save(ctx, logBackup); err != nil {
		log.GetLogger(ctx).Error("save new log backup error", zap.Error(err))
		return nil, err
	}
	return logBackup, nil
}

func (b *pitBackupModel) runEnablePitBackupDag(
	ctx context.Context,
	req *domain.AsyncCreateBackupReq,
	logBackup *entity.LogBackup,
	pitBackupSwitch *entity.PitBackupSwitch,
) (string, error) {
	tags := map[string]string{
		workflow.BusinessID:        strconv.FormatUint(logBackup.ID.Uint64(), 10),
		workflow.BusinessTypeKey:   string(workflow.BusinessTypePitBackup),
		workflow.BusinessActionKey: string(workflow.BusinessActionEnablePitBackup),
		workflow.ClusterID:         strconv.FormatUint(uint64(pitBackupSwitch.ClusterID), 10),
	}
	reqBytes, err := json.Marshal(req)
	if err != nil {
		log.GetLogger(ctx).Error("marshal req error", zap.Error(err))
		return "", err
	}
	vars := map[string]string{
		workflow.KeyPitBackupSwitchID:    strconv.FormatUint(pitBackupSwitch.ID.Uint64(), 10),
		workflow.KeyEnableLogBackupID:    strconv.FormatUint(logBackup.ID.Uint64(), 10),
		workflow.KeyAsyncCreateBackupReq: string(reqBytes),
		workflow.ClusterID:               strconv.FormatUint(uint64(pitBackupSwitch.ClusterID), 10),
	}
	res, err := b.workflowCli.RunDagWithTags(workflow.DagEnablePitBackup, vars, tags)
	if err != nil {
		log.GetLogger(ctx).Error("create dag failed", zap.Error(err))
		return "", err
	}
	return res.ID, nil
}

func (b *pitBackupModel) restartPitBackup(
	ctx context.Context,
	pitBackupSwitch *entity.PitBackupSwitch,
	cluster *mgmt.Cluster,
	req *domain.AsyncCreateBackupReq) (
	*domain.AsyncCreateBackupResp, bizerr.BizError,
) {
	if err := b.preCheckRestartPitBackup(ctx, pitBackupSwitch, cluster, req); err != nil {
		return nil, err
	}
	if req.DryRun {
		log.GetLogger(ctx).Info("dry run restart pit backup success", zap.Any("req", req))
		return nil, nil
	}

	resp := &domain.AsyncCreateBackupResp{}
	reqBytes, err := json.Marshal(req)
	if err != nil {
		log.GetLogger(ctx).Error("marshal req error", zap.Error(err))
		return resp, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}

	txF := func(ctx context.Context) error {
		nextStatus := b.switchStatusMatrix.CalculateNextStatusForAction(status_matrix.CurrentStatus{
			Status:  string(pitBackupSwitch.Status),
			Actions: pitBackupSwitch.Action.CurrentActions,
		}, string(entity.PitBackupSwitchActionDoRestart))
		pitBackupSwitch.Action.CurrentActions = nextStatus.Actions
		pitBackupSwitch.Status = entity.PitBackupSwitchStatus(nextStatus.Status)

		res, logBackupToCreate, err := b.startRestartPitBackupDag(ctx, cluster, pitBackupSwitch, reqBytes)
		if err != nil {
			return err
		}
		resp.WorkflowID = res.ID
		resp.BackupScheduleID = logBackupToCreate.BackupScheduleID.Uint64()
		if err := b.pitBackupSwitchRepo.Save(ctx, pitBackupSwitch); err != nil {
			log.GetLogger(ctx).Error("save pit backup switch failed", zap.Error(err))
			return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
		}
		return nil
	}
	if err := b.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		return txF(ctx)
	}); err != nil {
		log.GetLogger(ctx).Error("restart pit backup failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	return resp, nil
}

func (b *pitBackupModel) preCheckRestartPitBackup(
	ctx context.Context, pitBackupSwitch *entity.PitBackupSwitch,
	cluster *mgmt.Cluster, req *domain.AsyncCreateBackupReq,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	// check pit backup switch status
	isFriendly := b.switchStatusMatrix.IsActionFriendly2Status(status_matrix.CurrentStatus{
		Status:  string(pitBackupSwitch.Status),
		Actions: pitBackupSwitch.Action.CurrentActions,
	}, string(entity.PitBackupSwitchActionDoRestart))
	if !isFriendly {
		logger.Warn("restartPitBackup failed , status is not friendly",
			zap.Any("pit backup switch", pitBackupSwitch))
		return bizerr_wrapper.NewInvalidInputError(
			config.ErrActionParam, pitBackupSwitch.Status, entity.PitBackupSwitchActionDoEnable)
	}

	// check can create snapshot backup
	cloneReq, cErr := req.Clone()
	if cErr != nil {
		logger.Error("clone req error", zap.Error(cErr))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	cloneReq.DryRun = true
	_, err := b.snapshotBackupModel.AsyncCreate(ctx, cloneReq)
	if err != nil {
		return err
	}
	return nil
}

func (b *pitBackupModel) startRestartPitBackupDag(
	ctx context.Context,
	cluster *mgmt.Cluster,
	pitBackupSwitch *entity.PitBackupSwitch,
	reqBytes []byte,
) (*dagentity.DagInstance, *entity.LogBackup, error) {
	// get active log backup to run disable pit backup dag
	logBackupToStop, getErr := b.getActiveLogBackup(ctx, cluster)
	if getErr != nil {
		log.GetLogger(ctx).Error("get active log backup error", zap.Error(getErr))
		return nil, nil, getErr
	}

	logBackupToCreate, createErr := b.createLogBackupEntity(ctx, cluster)
	if createErr != nil {
		log.GetLogger(ctx).Error("create log backup error", zap.Error(createErr))
		return nil, nil, createErr
	}

	vars := map[string]string{
		workflow.KeyPitBackupSwitchID:    strconv.FormatUint(pitBackupSwitch.ID.Uint64(), 10),
		workflow.KeyDisableLogBackupID:   strconv.FormatUint(logBackupToStop.ID.Uint64(), 10),
		workflow.KeyEnableLogBackupID:    strconv.FormatUint(logBackupToCreate.ID.Uint64(), 10),
		workflow.KeyAsyncCreateBackupReq: string(reqBytes),
	}
	tags := map[string]string{
		workflow.BusinessID:        strconv.FormatUint(pitBackupSwitch.ID.Uint64(), 10),
		workflow.BusinessTypeKey:   string(workflow.BusinessTypePitBackup),
		workflow.BusinessActionKey: string(workflow.BusinessActionRestartPitBackup),
		workflow.ClusterID:         strconv.FormatUint(uint64(pitBackupSwitch.ClusterID), 10),
	}
	res, err := b.workflowCli.RunDagWithTags(workflow.DagRestartPitBackup, vars, tags)
	if err != nil {
		log.GetLogger(ctx).Error("create dag failed", zap.Error(err))
		return nil, nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	return res, logBackupToCreate, nil
}

//nolint:dupl
func (b *pitBackupModel) getActiveLogBackup(
	ctx context.Context, cluster *mgmt.Cluster,
) (*entity.LogBackup, bizerr.BizError) {
	filterF := func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
		qs = qs.ClusterIDEq(entity.ID(cluster.GetClusterId())).
			ProjectIDEq(entity.ID(cluster.GetProjectId())).
			TenantIDEq(entity.ID(cluster.GetOrganizationId())).
			DeletedAtIsNull().
			StatusNotIn(entity.LogBackupStatusStopped).
			PreloadBackupSchedule()
		return qs
	}
	logBackups, err := b.logBackupRepo.List(ctx, filterF)
	if err != nil {
		log.GetLogger(ctx).Error("list log backup error", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	if len(logBackups) > 1 {
		log.GetLogger(ctx).Error("more than one active log backup existed this should not happen", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	if len(logBackups) == 0 {
		log.GetLogger(ctx).Error("no active log backup existed this should not happen", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	return &logBackups[0], nil
}

func (b *pitBackupModel) AsyncDeleteLogBackup(
	ctx context.Context, req *domain.DeleteLogBackupReq,
) (*domain.DeleteLogBackupResp, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).With(zap.Any("log backup id", req.LogBackupID))

	logBackup, err := b.logBackupRepo.Load(ctx, req.LogBackupID)
	if err != nil {
		logger.Warn("load log backup error", zap.Error(err))
		return nil, bizerr_wrapper.NewNotFoundErr(config.ErrMsgNotFound, req.LogBackupID)
	}

	// handle deleted backup
	if logBackup.DeletedAt != nil {
		logger.Warn("log backup already deleted")
		return &domain.DeleteLogBackupResp{}, nil
	}

	if err := b.preCheckForAsyncDelete(ctx, logBackup); err != nil {
		return nil, err
	}

	// delete ops
	resp := &domain.DeleteLogBackupResp{}
	txF := func(ctx context.Context) error {
		deletedTime := time.Now()
		if err := b.deleteLogBackupEntity(ctx, &deletedTime, logBackup); err != nil {
			logger.Error("failed deleteLogBackupEntity", zap.Error(err))
			return err
		}

		if workflowID, err := b.createLogBackupDeleteDag(ctx, logger, logBackup); err != nil {
			logger.Error("failed to create delete backup dag", zap.Error(err))
			return err
		} else {
			resp.WorkflowID = workflowID
		}
		return nil
	}

	if err := b.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		return txF(ctx)
	}); err != nil {
		logger.Error("delete log backup failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, logBackup.ClusterID)
	}
	return resp, nil
}

func (b *pitBackupModel) preCheckForAsyncDelete(ctx context.Context, logBackup *entity.LogBackup) bizerr.BizError {
	return b.checkRunningRestore(ctx, logBackup)
}

func (b *pitBackupModel) checkRunningRestore(ctx context.Context, logBackup *entity.LogBackup) bizerr.BizError {
	queryF := func(qs entity.RestoreQuerySet) entity.RestoreQuerySet {
		return qs.LogBackupIDEq(logBackup.ID).
			StatusIn(entity.RestoreStatusPENDING, entity.RestoreStatusRUNNING).
			OrderAscByCreatedAt()
	}
	total, err := b.restoreRepo.Count(ctx, queryF)
	if err != nil {
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, logBackup.ClusterID)
	}
	if total > 0 {
		return bizerr_wrapper.NewInvalidInputError(
			config.ErrMsgAsyncDeleteBackupPreCheckRunningRestore, logBackup.ID)
	}
	return nil
}

func (b *pitBackupModel) deleteLogBackupEntity(
	ctx context.Context, deletedTime *time.Time, logBackup *entity.LogBackup,
) bizerr.BizError {
	logBackup.DeletedAt = deletedTime
	if err := b.logBackupRepo.Save(ctx, logBackup); err != nil {
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, logBackup.ClusterID)
	}
	return nil
}

func (b *pitBackupModel) createLogBackupDeleteDag(
	ctx context.Context, logger *zap.Logger, logBackup *entity.LogBackup,
) (string, bizerr.BizError) {
	vars := map[string]string{
		workflow.KeyLogBackupID: strconv.FormatUint(uint64(logBackup.ID), 10),
	}
	tags := map[string]string{
		workflow.BusinessID:        strconv.FormatUint(uint64(logBackup.ID), 10),
		workflow.BusinessTypeKey:   string(workflow.BusinessTypeLogBackup),
		workflow.BusinessActionKey: string(workflow.BusinessActionLogBackupDelete),
		workflow.ClusterID:         strconv.FormatUint(uint64(logBackup.ClusterID), 10),
	}
	res, err := b.workflowCli.RunDagWithTags(workflow.DagDeleteLogBackup, vars, tags)
	if err != nil {
		return "", bizerr_wrapper.NewInternalErr("failed to create delete log backup dag")
	}
	return res.ID, nil
}

func (b *pitBackupModel) ListPiTRTimeRanges(
	ctx context.Context, req *domain.PiTRTimeRangeListReq,
) (*domain.PiTRTimeRangeListResp, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	ctx = log.WithTraceID(ctx)

	backups, logBackups, getErr := b.getBackupsAndLogBackupsByCluster(ctx, req.ClusterId)
	if getErr != nil {
		return nil, getErr
	}

	logBackupTimeRanges := b.genPiTRTimeRanges(ctx, backups, logBackups)
	// no restore time range, we need to give user tips
	if len(logBackupTimeRanges) == 0 {
		tips, err := b.genPiTRTips(ctx, backups, logBackups)
		if err != nil {
			logger.Error("failed to genPiTRTips",
				zap.Uint64("clusterID", req.ClusterId), zap.Error(err))
			return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterId)
		}
		return &domain.PiTRTimeRangeListResp{
			Msg: tips,
		}, nil
	}

	regionTimeRanges, err := b.mergePiTRTimeRanges(ctx, req.ClusterId, logBackupTimeRanges)
	if err != nil {
		logger.Error("failed to group PiTR time ranges by region",
			zap.Uint64("clusterID", req.ClusterId), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterId)
	}

	resp := &domain.PiTRTimeRangeListResp{
		Items: regionTimeRanges,
	}
	return resp, nil
}

func (b *pitBackupModel) getBackupsAndLogBackupsByCluster(
	ctx context.Context, clusterId uint64,
) ([]entity.Backup, []entity.LogBackup, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	logBackups, err := b.listLogBackupsByCluster(ctx, clusterId)
	if err != nil {
		logger.Error("failed to list log backups",
			zap.Uint64("clusterID", clusterId), zap.Error(err))
		return nil, nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, clusterId)
	}

	backups, err := b.listBackupsByCluster(ctx, clusterId)
	if err != nil {
		logger.Error("failed to list log backups",
			zap.Uint64("clusterID", clusterId), zap.Error(err))
		return nil, nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, clusterId)
	}

	return backups, logBackups, nil
}

func (b *pitBackupModel) listLogBackupsByCluster(
	ctx context.Context, clusterId uint64,
) ([]entity.LogBackup, error) {
	filterF := func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
		qs = qs.ClusterIDEq(entity.ID(clusterId)).
			DeletedAtIsNull().
			StatusNotIn(entity.LogBackupStatusStopped).
			PreloadBackupSchedule().
			OrderDescByCreatedAt()
		return qs
	}
	return b.logBackupRepo.List(ctx, filterF)
}

func (b *pitBackupModel) listBackupsByCluster(
	ctx context.Context, clusterId uint64,
) ([]entity.Backup, error) {
	filterF := func(qs entity.BackupQuerySet) entity.BackupQuerySet {
		qs = qs.ClusterIDEq(entity.ID(clusterId)).
			DeletedAtIsNull().
			StatusNotIn(entity.BackupStatusFailed).
			PreloadBackupSchedule().
			OrderDescByCreatedAt()
		return qs
	}

	return b.backupRepo.ListExternal(ctx, filterF)
}

func (b *pitBackupModel) genPiTRTimeRanges(
	ctx context.Context, backups []entity.Backup, logBackups []entity.LogBackup,
) []*domain.PiTRTimeRange {
	logBackupTimeRanges := make([]*domain.PiTRTimeRange, 0, len(logBackups))
	for _, logBackup := range logBackups {
		startTs, endTs, found := logBackup.GenPiTRTimeRange(ctx, backups)
		if !found {
			continue
		}
		timeRange := &domain.PiTRTimeRange{
			LogBackup:   logBackup,
			LogBackupID: logBackup.ID,
			BucketID:    logBackup.BucketID,
			StartTS:     startTs,
			EndTS:       endTs,
		}

		logBackupTimeRanges = append(logBackupTimeRanges, timeRange)
	}

	return logBackupTimeRanges
}

// genPiTRTips generate PiTR Tips when there is no restore ranges.
func (b *pitBackupModel) genPiTRTips(
	ctx context.Context, backups []entity.Backup, logBackups []entity.LogBackup,
) (string, error) {
	// no log backup
	if len(logBackups) == 0 {
		return offTip, nil
	}

	current := logBackups[0]

	// log backup failed
	if current.Status == entity.LogBackupStatusFailed {
		return errTip, nil
	}

	// handle region switch which will stop and start log backup and will affect log backup status
	// region switch and log backup scheduled are both will take effect during next backup, we can give the same tips
	if current.Status == entity.LogBackupStatusScheduled {
		return b.genTipsWhenScheduledToNextBackup(current, backups), nil
	}

	// log backup stopped
	if current.Status == entity.LogBackupStatusStopping || current.Status == entity.LogBackupStatusStopped {
		return offTip, nil
	}

	// others status, we can give the tips when log backup is running
	return b.genTipsWhenLogBackupRunning(current, backups)
}

func (b *pitBackupModel) genTipsWhenLogBackupRunning(
	current entity.LogBackup, backups []entity.Backup,
) (string, error) {
	_, currentCheckPointTsP, err := current.GenLogBackupValidRange()
	if err != nil {
		return "", err
	}
	// no checkpoint ts, should wait log backup checkpoint ts update.
	if currentCheckPointTsP == nil {
		return catchingTip, nil
	}

	// calculate next backup ts which is greater than log backup checkpoint ts
	var backupTsP *int64
	for i := 0; i < len(backups); i++ {
		if backups[i].Status != entity.BackupStatusSuccess {
			continue
		}
		// we may switch bucket, when execute PiTR, full backup and log backup need be stored in same bucket
		if backups[i].BucketID != current.BucketID {
			continue
		}

		backupTS, err := util.ParseTSFromTSString(backups[i].StartTimeStamp)
		if err != nil {
			continue
		}

		if backupTS < *currentCheckPointTsP {
			break
		}
		backupTsP = &backupTS
	}

	// no next backup, maybe latest backup is running, otherwise, should make a new backup
	if backupTsP == nil {
		// latest backup is running
		return b.genBackupTip(backups), nil
	}
	return catchingTip, nil
}

func (b *pitBackupModel) genBackupTip(backups []entity.Backup) string {
	if len(backups) > 0 &&
		(backups[0].Status == entity.BackupStatusPending || backups[0].Status == entity.BackupStatusRunning) {
		if backups[0].BackupSchedule != nil {
			return fmt.Sprintf(backupRunningTip, backups[0].BackupSchedule.Name)
		}
		return fmt.Sprintf(backupRunningTip, "")
	}
	return noBackupTip
}

func (b *pitBackupModel) genTipsWhenScheduledToNextBackup(current entity.LogBackup, backups []entity.Backup) string {
	getNextScheduleTime := func() string {
		if current.BackupSchedule == nil ||
			current.BackupSchedule.NextBackupTime == nil ||
			current.BackupSchedule.Schedule == "" {
			return ""
		}
		expr := cronexpr.MustParse(current.BackupSchedule.Schedule)
		nextTime := expr.Next(time.Now())
		return nextTime.UTC().Format("2006-01-02 15:04:05 MST")
	}

	// scheduled will take effect during next backup
	creatAt := current.CreatedAt.Unix()
	if len(backups) > 0 && backups[0].CreatedAt.Unix() > creatAt &&
		(backups[0].Status == entity.BackupStatusPending || backups[0].Status == entity.BackupStatusRunning) {
		if backups[0].BackupSchedule != nil {
			return fmt.Sprintf(backupRunningTip, backups[0].BackupSchedule.Name)
		}
		return fmt.Sprintf(backupRunningTip, "")
	}
	return fmt.Sprintf(scheduledTip, getNextScheduleTime())
}

func (b *pitBackupModel) mergePiTRTimeRanges(
	ctx context.Context, clusterID uint64, logBackupTimeRanges []*domain.PiTRTimeRange,
) ([]*domain.PiTRRegionTimeRange, error) {
	cluster, err := b.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, clusterID)
	if err != nil {
		return nil, errors.Annotate(err, "get cluster")
	}

	cloudRegions, err := b.coreCli.ListCloudRegions(
		ctx, &core.ListCloudRegionsReq{
			ArchRouting: &core.ArchRouting{ProjectId: cluster.GetProjectId()},
			Filter:      &core.CloudRegionFilter{CloudProviderNames: []string{cluster.GetCloudProvider()}},
		})
	if err != nil {
		return nil, errors.Annotate(err, "get cloud regions")
	}

	var regionTimeRanges []*domain.PiTRRegionTimeRange
	var groupErr error
	if override, ok := internal_config.Global().BackupConfig.CRRFOverrides.GetMatched(
		cluster.GetOrganizationId(), cluster.GetProjectId(), cluster.GetClusterId(),
		cluster.GetCloudProvider(), cluster.GetRegion(),
	); ok && override.CalculatePiTRTimeRangeV2 {
		regionTimeRanges, groupErr = b.groupPiTRTimeRangesByRegionV2(ctx, logBackupTimeRanges, cluster, cloudRegions)
	} else {
		regionTimeRanges, groupErr = b.groupPiTRTimeRangesByRegion(ctx, logBackupTimeRanges, cluster, cloudRegions)
	}
	if groupErr != nil {
		return nil, groupErr
	}

	for _, regionTimeRange := range regionTimeRanges {
		regionTimeRange.TimeRanges = b.mergeSingleRegionPiTRTimeRanges(ctx, regionTimeRange.TimeRanges)
	}
	return regionTimeRanges, nil
}

// because the bucket switching, some log backups are stored in old bucket and others are stored in new bucket
// the buckets may locate in different regions
// we should group PiTR time ranges of log backups by the region they stored, every region has their own time ranges

//nolint:funlen
func (b *pitBackupModel) groupPiTRTimeRangesByRegion(
	ctx context.Context, logBackupTimeRanges []*domain.PiTRTimeRange,
	cluster *mgmt.Cluster, cloudRegions map[string]map[string]*core.CloudRegion,
) ([]*domain.PiTRRegionTimeRange, error) {
	dc := cluster.GetDedicatedCluster()
	regions := cloudRegions[cluster.GetCloudProvider()]

	clusterRegionTimeRange := &domain.PiTRRegionTimeRange{
		Region: &domain.RegionInfo{
			Name:        cluster.GetRegion(),
			DisplayName: dc.GetRegionDisplayName(),
		},
		TimeRanges: make([]*domain.PiTRTimestampRange, 0, len(logBackupTimeRanges)),
	}
	const rangeSize = 2 // Start and End
	timeRanges := make([]*domain.PiTRRegionTimeRange, 0, rangeSize)
	timeRanges = append(timeRanges, clusterRegionTimeRange)
	timeRangeMap := make(map[string]*domain.PiTRRegionTimeRange)
	timeRangeMap[cluster.GetRegion()] = clusterRegionTimeRange
	clusterRegions := []string{cluster.GetRegion()}
	for _, logBackupTimeRange := range logBackupTimeRanges {
		// use cluster region as default regions, if there is a dual region bucket, use the regions of bucket instead
		logBackupRegions, err := b.queryLogBackupRegions(ctx, clusterRegions, logBackupTimeRange)
		if err != nil {
			return nil, err
		}
		// put this time range to every region
		for _, region := range logBackupRegions {
			timeRange, ok := timeRangeMap[region]
			if !ok {
				regionInfo := &domain.RegionInfo{
					Name:        region,
					DisplayName: region,
				}
				if region, ok := regions[region]; ok {
					regionInfo.DisplayName = region.DisplayName
				}
				timeRange = &domain.PiTRRegionTimeRange{
					Region:     regionInfo,
					TimeRanges: make([]*domain.PiTRTimestampRange, 0, len(logBackupTimeRanges)),
				}
				timeRangeMap[region] = timeRange
			}

			timeRange.TimeRanges = append(timeRange.TimeRanges, &domain.PiTRTimestampRange{
				StartTS: timestamppb.New(time.Unix(logBackupTimeRange.StartTS, 0)),
				EndTS:   timestamppb.New(time.Unix(logBackupTimeRange.EndTS, 0)),
			})
		}
	}

	for region, regionTimeRange := range timeRangeMap {
		if region == cluster.GetRegion() {
			continue
		}
		timeRanges = append(timeRanges, regionTimeRange)
	}
	return timeRanges, nil
}

func (b *pitBackupModel) groupPiTRTimeRangesByRegionV2(
	ctx context.Context, logBackupTimeRanges []*domain.PiTRTimeRange,
	cluster *mgmt.Cluster, cloudRegions map[string]map[string]*core.CloudRegion,
) ([]*domain.PiTRRegionTimeRange, error) {
	logger := log.GetLogger(ctx)
	regions := cloudRegions[cluster.GetCloudProvider()]

	timeRangeMap := make(map[string]*domain.PiTRRegionTimeRange)
	for _, logBackupTimeRange := range logBackupTimeRanges {
		logBackup := logBackupTimeRange.LogBackup

		// If there is no bucket snapshot (e.g. legacy rows), fall back to the
		// cluster's primary region with the main end timestamp, mirroring V1 behavior.
		if logBackup.BucketSettingSnapshot == nil {
			b.appendRegionTimeRange(timeRangeMap, regions, cluster.GetRegion(), logBackupTimeRange.StartTS, logBackupTimeRange.EndTS)
			continue
		}

		for _, bucket := range logBackup.BucketSettingSnapshot.Buckets {
			for _, region := range bucket.Regions {
				var endTs int64
				if bucket.Master {
					// Primary bucket: all regions use the main checkpoint end timestamp.
					endTs = logBackupTimeRange.EndTS
				} else {
					// Secondary bucket: use the per-region CRPiTR checkpoint.
					// On any failure (error or file not found), fall back to the
					// primary end timestamp so the region is still surfaced to callers.
					checkpoint, found, err := b.getCheckpointForCRPiTR(ctx, logBackup, region, b.getStorageFunc)
					if err != nil {
						logger.Warn("get checkpoint for CR PiTR, falling back to primary endTS",
							zap.String("region", region), zap.Error(err))
						endTs = logBackupTimeRange.EndTS
					} else if !found {
						endTs = logBackupTimeRange.EndTS
					} else {
						// If the replica checkpoint lags behind the range start, the region
						// has no recoverable data in this interval — skip it entirely.
						if checkpoint < logBackupTimeRange.StartTS {
							logger.Warn("skip secondary region: checkpoint before range start",
								zap.String("region", region),
								zap.Int64("startTS", logBackupTimeRange.StartTS),
								zap.Int64("checkpointTS", checkpoint))
							continue
						}
						endTs = checkpoint
					}
				}
				b.appendRegionTimeRange(timeRangeMap, regions, region, logBackupTimeRange.StartTS, endTs)
			}
		}
	}
	var timeRanges []*domain.PiTRRegionTimeRange
	for _, timeRange := range timeRangeMap {
		timeRanges = append(timeRanges, timeRange)
	}
	return timeRanges, nil
}

func (b *pitBackupModel) appendRegionTimeRange(
	timeRangeMap map[string]*domain.PiTRRegionTimeRange,
	regions map[string]*core.CloudRegion,
	region string,
	startTS, endTs int64,
) {
	timeRange, ok := timeRangeMap[region]
	if !ok {
		regionInfo := &domain.RegionInfo{
			Name:        region,
			DisplayName: region,
		}
		if r, ok := regions[region]; ok {
			regionInfo.DisplayName = r.DisplayName
		}
		timeRange = &domain.PiTRRegionTimeRange{Region: regionInfo}
		timeRangeMap[region] = timeRange
	}
	timeRange.TimeRanges = append(timeRange.TimeRanges, &domain.PiTRTimestampRange{
		StartTS: timestamppb.New(time.Unix(startTS, 0)),
		EndTS:   timestamppb.New(time.Unix(endTs, 0)),
	})
}

func (b *pitBackupModel) queryLogBackupRegions(
	ctx context.Context, clusterRegions []string, logBackupTimeRange *domain.PiTRTimeRange,
) ([]string, error) {
	if logBackupTimeRange.LogBackup.BucketSettingSnapshot != nil {
		var logBackupRegions []string
		pie.Of(logBackupTimeRange.LogBackup.BucketSettingSnapshot.Buckets).Each(func(t entity.BackupSettingBucket) {
			logBackupRegions = append(logBackupRegions, t.Regions...)
		})
		return logBackupRegions, nil
	}
	return clusterRegions, nil
}

// the timeRanges param is the time ranges from later to earlier, we should merge the successive ranges.
// for example, we have ranges [2022-09-25 15:04:05, 2022-09-30 18:08:09], [2022-09-10 14:06:07, 2022-09-25 15:04:05], [2022-09-01 12:09:10, 2022-09-09 13:23:56]
// we will merge to [2022-09-10 14:06:07, 2022-09-30 18:08:09], [2022-09-01 12:09:10, 2022-09-09 13:23:56]
func (b *pitBackupModel) mergeSingleRegionPiTRTimeRanges(
	ctx context.Context, timeRanges []*domain.PiTRTimestampRange,
) []*domain.PiTRTimestampRange {
	if len(timeRanges) == 0 {
		return []*domain.PiTRTimestampRange{}
	}

	pitrTimeRanges := make([]*domain.PiTRTimestampRange, 0, len(timeRanges))
	pitrTimeRange := timeRanges[0]

	for i := 1; i < len(timeRanges); i++ {
		if pitrTimeRange.StartTS.GetSeconds() <= timeRanges[i].EndTS.GetSeconds() &&
			pitrTimeRange.StartTS.GetSeconds() > timeRanges[i].StartTS.GetSeconds() {
			pitrTimeRange.StartTS = timeRanges[i].StartTS
			continue
		}

		pitrTimeRanges = append(pitrTimeRanges, pitrTimeRange)
		pitrTimeRange = timeRanges[i]
	}

	pitrTimeRanges = append(pitrTimeRanges, pitrTimeRange)
	return pitrTimeRanges
}