package impl

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"github.com/hashicorp/go-multierror"
	"github.com/tidbcloud/dataflow-service/app/internal/feature_gate"
	"github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	backup_repo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	service_config "github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
)

const (
	// DefaultCompactBackupCycle is the default compact backup cycle duration
	DefaultCompactBackupCycle = 5*time.Hour + 50*time.Minute
)

type CompactBackupSchedulerImpl struct {
	transProvider      db.TransProvider
	logBackupRepo      backup_repo.LogBackupRepo
	compactBackupRepo  backup_repo.CompactBackupRepo
	compactBackupModel backup.CompactBackupModel
	compactFacade      infraapi.CompactFacade
	mgmtCli            grpcli.MgmtSvcGrpcClient
	featureGate        feature_gate.FeatureGate
}

var _ backup.CompactBackupScheduler = (*CompactBackupSchedulerImpl)(nil)

func NewCompactBackupScheduler(
	transProvider db.TransProvider,
	logBackupRepo backup_repo.LogBackupRepo,
	compactBackupRepo backup_repo.CompactBackupRepo,
	compactBackupModel backup.CompactBackupModel,
	compactFacade infraapi.CompactFacade,
	mgmtCli grpcli.MgmtSvcGrpcClient,
	featureGate feature_gate.FeatureGate,
) backup.CompactBackupScheduler {
	return &CompactBackupSchedulerImpl{
		transProvider:      transProvider,
		logBackupRepo:      logBackupRepo,
		compactBackupRepo:  compactBackupRepo,
		compactBackupModel: compactBackupModel,
		compactFacade:      compactFacade,
		mgmtCli:            mgmtCli,
		featureGate:        featureGate,
	}
}

func (s *CompactBackupSchedulerImpl) CompactLogBackup(ctx context.Context) bizerr.BizError {
	ctx = log.WithLogger(ctx, log.GetLogger(ctx).With(zap.String("worker", "compact-log-backup")))
	logger := log.GetLogger(ctx)
	logBackups, err := s.getRunningLogBackups(ctx)
	if err != nil {
		logger.Error("failed to get running log backups", zap.Error(err))
		return err
	}

	if len(logBackups) == 0 {
		return nil
	}
	logger.Info("found running log backups for compact check",
		zap.Int("count", len(logBackups)))
	var multiErr error
	successCount, failCount := 0, 0
	for _, logBackup := range logBackups {
		logger := logger.With(
			zap.Uint64("log_backup_id", logBackup.ID.Uint64()),
			zap.Uint64("cluster_id", logBackup.ClusterID.Uint64()),
			zap.String("log_backup_status", string(logBackup.Status)),
			zap.String("checkpoint_ts", logBackup.CheckpointTS),
		)
		if err := s.processLogBackup(ctx, logger, &logBackup); err != nil {
			logger.Error("process log backup failed",
				zap.Uint64("cluster_id", logBackup.ClusterID.Uint64()),
				zap.Uint64("tenant_id", logBackup.TenantID.Uint64()),
				zap.Uint64("project_id", logBackup.ProjectID.Uint64()),
				zap.Error(err))
			multiErr = multierror.Append(multiErr, err)
			failCount++
		} else {
			successCount++
		}
	}

	logger.Info("compact check completed", zap.Int("total", len(logBackups)),
		zap.Int("success", successCount), zap.Int("fail", failCount))

	if multiErr != nil {
		return bizerr.WrapError(multiErr, "compact log backup failed")
	}
	return nil
}

func (s *CompactBackupSchedulerImpl) getRunningLogBackups(ctx context.Context) ([]entity.LogBackup, bizerr.BizError) {
	logBackups, err := s.logBackupRepo.List(ctx, func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
		return qs.StatusEq(entity.LogBackupStatusRunning).
			CheckpointTSNe("").
			DeletedAtIsNull().
			OrderAscByCreatedAt()
	})

	return logBackups, err
}

func (s *CompactBackupSchedulerImpl) processLogBackup(
	ctx context.Context,
	logger *zap.Logger,
	logBackup *entity.LogBackup,
) bizerr.BizError {
	// Check if compact log backup feature is enabled for this cluster
	enabled, err := s.checkCompactFeatureEnabled(ctx, logger, logBackup)
	if err != nil {
		return err
	}
	if !enabled {
		return nil
	}

	shouldCompact, err := s.ShouldCompactLogBackup(ctx, logBackup)
	if err != nil {
		logger.Error("failed to check if log backup should be compacted", zap.Error(err))
		return err
	}
	if !shouldCompact {
		startTS := "nil"
		if logBackup.StartTS != nil {
			startTS = *logBackup.StartTS
		}
		logger.Info("skip compact for this cycle",
			zap.String("reason", "conditions not met"), zap.String("start_ts", startTS))
		return nil
	}

	// Check if there's already a running compact task for this log backup
	hasRunning, err := s.hasRunningCompactTask(ctx, logBackup)
	if err != nil {
		logger.Error("failed to check for running compact tasks", zap.Error(err))
		return err
	}

	if hasRunning {
		logger.Info("skipping compact task creation, already have running task")
		return nil
	}

	if err := s.createCompactTask(ctx, logger, logBackup); err != nil {
		logger.Error("failed to create compact task", zap.Error(err))
		return err
	}

	return nil
}

// getCompactCycle retrieves the compact cycle duration for the given cluster
func (s *CompactBackupSchedulerImpl) getCompactCycle(
	ctx context.Context, clusterID uint64, logger *zap.Logger,
) time.Duration {
	cluster, err := s.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, clusterID)
	if err != nil {
		logger.Error("failed to get cluster info for compact cycle",
			zap.Uint64("cluster_id", clusterID), zap.Error(err))
		// Fall back to default cycle if cluster info is not available
		compactCycle, parseErr := service_config.Global().CompactConfig.GetCompactCycleDuration(
			0, 0, 0, "", // Use global defaults by passing zero/empty values
		)
		if parseErr != nil {
			logger.Error("failed to parse default compact backup cycle", zap.Error(parseErr))
			return DefaultCompactBackupCycle
		}
		return compactCycle
	}

	// Use cluster-specific compact cycle configuration
	compactCycle, parseErr := service_config.Global().CompactConfig.GetCompactCycleDuration(
		cluster.GetClusterId(),
		cluster.GetOrganizationId(),
		cluster.GetProjectId(),
		cluster.GetRegion(),
	)
	if parseErr != nil {
		logger.Error("failed to parse cluster-specific compact backup cycle", zap.Error(parseErr))
		return DefaultCompactBackupCycle
	}
	return compactCycle
}

// checkCompactConditions verifies if the log backup meets compact conditions
func (s *CompactBackupSchedulerImpl) checkCompactConditions(
	ctx context.Context,
	logBackup *entity.LogBackup,
	compactCycle time.Duration,
	logger *zap.Logger,
) bool {
	// 1. Check if the log backup has run long enough since its start time
	runTimeEnough := s.CheckLogBackupRunTimeEnough(ctx, logBackup, compactCycle)
	if !runTimeEnough {
		logger.Info("log backup has not run long enough, skip compact")
		return false
	}

	// 2. Check if we reach the compact cycle
	reachCompactCycle := s.CheckWhetherReachCompactCycle(ctx, logBackup, compactCycle)
	logger.Info("checked compact cycle",
		zap.Bool("reach_compact_cycle", reachCompactCycle),
		zap.Duration("compact_cycle", compactCycle))

	if !reachCompactCycle {
		logger.Debug("compact cycle not reached, skip compact")
		return false
	}

	logger.Info("log backup should be compacted",
		zap.Bool("run_time_enough", runTimeEnough),
		zap.Bool("reach_compact_cycle", reachCompactCycle))

	return true
}

func (s *CompactBackupSchedulerImpl) ShouldCompactLogBackup(
	ctx context.Context,
	logBackup *entity.LogBackup,
) (bool, bizerr.BizError) {
	logger := log.GetLogger(ctx).With(
		zap.Uint64("log_backup_id", logBackup.ID.Uint64()),
		zap.String("status", string(logBackup.Status)),
		zap.String("checkpoint_ts", logBackup.CheckpointTS),
	)

	if logBackup.Status != entity.LogBackupStatusRunning {
		logger.Debug("log backup is not running, skip compact",
			zap.String("current_status", string(logBackup.Status)))
		return false, nil
	}

	if logBackup.CheckpointTS == "" {
		logger.Info("log backup has empty checkpoint_ts, skip compact")
		return false, nil
	}

	compactCycle := s.getCompactCycle(ctx, logBackup.ClusterID.Uint64(), logger)
	shouldCompact := s.checkCompactConditions(ctx, logBackup, compactCycle, logger)

	return shouldCompact, nil
}

func (s *CompactBackupSchedulerImpl) CheckLogBackupRunTimeEnough(
	ctx context.Context,
	logBackup *entity.LogBackup,
	compactCycle time.Duration,
) bool {
	logger := log.GetLogger(ctx)

	if logBackup.StartTS == nil || *logBackup.StartTS == "" {
		logger.Debug("log backup has no start_ts",
			zap.Bool("start_ts_nil", logBackup.StartTS == nil),
			zap.String("start_ts_value", func() string {
				if logBackup.StartTS != nil {
					return *logBackup.StartTS
				}
				return "nil"
			}()))
		return false
	}

	startTime, err := s.parseTS(*logBackup.StartTS)
	if err != nil {
		logger.Warn("failed to parse LogBackup startTS",
			zap.String("start_ts", *logBackup.StartTS),
			zap.Error(err))
		return false
	}

	runTime := time.Since(startTime)
	enough := runTime >= compactCycle

	logger.Debug("checked log backup run time",
		zap.Time("start_time", startTime),
		zap.Duration("run_time", runTime),
		zap.Duration("required_time", compactCycle),
		zap.Bool("enough", enough))

	return enough
}

func (s *CompactBackupSchedulerImpl) CheckWhetherReachCompactCycle(
	ctx context.Context,
	logBackup *entity.LogBackup,
	compactCycle time.Duration,
) bool {
	logger := log.GetLogger(ctx)

	// TODO: what if the tasks keep failing or running?
	lastSuccessTask, err := s.getLastSuccessCompactTask(ctx, logBackup.ID)
	if err != nil {
		logger.Warn("failed to get last success compact task", zap.Error(err))
		return false
	}

	if lastSuccessTask == nil {
		logger.Debug("no previous successful compact task found, should compact")
		return true
	}

	timeSinceLastCompact := time.Since(lastSuccessTask.CreatedAt)
	reachCycle := timeSinceLastCompact >= compactCycle

	logger.Debug("checked compact cycle since last success",
		zap.Time("last_compact_time", lastSuccessTask.CreatedAt),
		zap.Uint64("last_compact_id", lastSuccessTask.ID.Uint64()),
		zap.Duration("time_since_last", timeSinceLastCompact),
		zap.Duration("required_cycle", compactCycle),
		zap.Bool("reach_cycle", reachCycle))

	return reachCycle
}

func (s *CompactBackupSchedulerImpl) parseTS(tsStr string) (time.Time, error) {
	tso, err := strconv.ParseUint(tsStr, 10, 64)
	if err != nil {
		return time.Time{}, err
	}

	return utils.GetTimeFromTS(tso), nil
}

func (s *CompactBackupSchedulerImpl) getLastSuccessCompactTask(
	ctx context.Context,
	logBackupID entity.ID,
) (*entity.CompactBackup, bizerr.BizError) {
	logBackupIDVal := logBackupID.Uint64()
	// Maybe also the failed ones
	statusVal := entity.CompactBackupStatusSuccess

	filter := &backup_repo.CompactBackupFilter{
		LogBackupID: &logBackupIDVal,
		Status:      &statusVal,
	}

	compactTasks, err := s.compactBackupRepo.ListByFilter(ctx, filter)
	if err != nil {
		return nil, err
	}

	if len(compactTasks) == 0 {
		return nil, nil
	}

	return compactTasks[0], nil
}

// checkCompactFeatureEnabled checks if compact log backup feature is enabled for the cluster
func (s *CompactBackupSchedulerImpl) checkCompactFeatureEnabled(
	ctx context.Context, logger *zap.Logger, logBackup *entity.LogBackup,
) (bool, bizerr.BizError) {
	// Get cluster information for provider and region
	cluster, err := s.mgmtCli.GetDedicatedClusterBasicByID(ctx, logBackup.ClusterID.Uint64())
	if err != nil {
		logger.Error("failed to get cluster information for feature gate check",
			zap.Uint64("cluster_id", logBackup.ClusterID.Uint64()), zap.Error(err))
		return false, err
	}

	// Check if compact log backup feature is enabled for this cluster
	featureGateReq := &feature_gate.QueryFeatureGateReq{
		TenantID:      logBackup.TenantID.Uint64(),
		ProjectID:     logBackup.ProjectID.Uint64(),
		ClusterID:     logBackup.ClusterID.Uint64(),
		Region:        cluster.GetRegion(),
		InfraProvider: cluster.GetCloudProvider(),
	}

	if !s.featureGate.IsFeatureEnabled(featureGateReq, feature_gate.CompactLogBackup) {
		logger.Info("CompactLogBackup feature is not enabled for this cluster, skipping",
			zap.Uint64("tenant_id", logBackup.TenantID.Uint64()),
			zap.Uint64("project_id", logBackup.ProjectID.Uint64()),
			zap.Uint64("cluster_id", logBackup.ClusterID.Uint64()),
			zap.String("region", cluster.GetRegion()),
			zap.String("provider", cluster.GetCloudProvider()))
		return false, nil
	}

	return true, nil
}

// getLocalRunningTasks retrieves all running and pending compact backup tasks from local DB
func (s *CompactBackupSchedulerImpl) getLocalRunningTasks(
	ctx context.Context, logBackupID uint64,
) ([]*entity.CompactBackup, bizerr.BizError) {
	runningStatus := entity.CompactBackupStatusRunning
	pendingStatus := entity.CompactBackupStatusPending

	// Check for running tasks
	filter := &backup_repo.CompactBackupFilter{
		LogBackupID: &logBackupID,
		Status:      &runningStatus,
	}
	runningTasks, err := s.compactBackupRepo.ListByFilter(ctx, filter)
	if err != nil {
		return nil, err
	}

	// Check for pending tasks
	filter.Status = &pendingStatus
	pendingTasks, err := s.compactBackupRepo.ListByFilter(ctx, filter)
	if err != nil {
		return nil, err
	}

	// Combine all tasks
	return append(runningTasks, pendingTasks...), nil
}

// getClusterRegion fetches cluster information and returns the region name
func (s *CompactBackupSchedulerImpl) getClusterRegion(
	ctx context.Context, clusterID uint64,
) (string, bizerr.BizError) {
	cluster, bizErr := s.mgmtCli.GetDedicatedClusterBasicByID(ctx, clusterID)
	if bizErr != nil {
		return "", bizErr
	}
	return cluster.GetRegion(), nil
}

// handleTaskWithoutDownstreamID handles tasks that have no downstream ID assigned
func (s *CompactBackupSchedulerImpl) handleTaskWithoutDownstreamID(
	ctx context.Context, task *entity.CompactBackup, logger *zap.Logger,
) bizerr.BizError {
	logger.Warn("task has no downstream ID, marking as failed")

	task.Status = entity.CompactBackupStatusFailed
	task.ErrMsg = "No downstream ID assigned - task creation likely failed"
	now := time.Now()
	task.FinishedAt = &now

	if updateErr := s.compactBackupRepo.Save(ctx, task); updateErr != nil {
		logger.Error("failed to update compact backup status to failed", zap.Error(updateErr))
		return updateErr
	}
	return nil
}

// updateTaskStatus updates task status based on downstream response
func (s *CompactBackupSchedulerImpl) updateTaskStatus(
	ctx context.Context, task *entity.CompactBackup, resp *infraapi.CompactBackupResponse,
	logger *zap.Logger,
) bizerr.BizError {
	logger.Info("[DOWNSTREAM_CHECK] downstream task is finished, updating local status",
		zap.String("downstream_status", string(resp.Status)))

	if resp.Status == infraapi.CompactBackupStatusComplete {
		task.Status = entity.CompactBackupStatusSuccess
	} else {
		task.Status = entity.CompactBackupStatusFailed
		task.ErrMsg = resp.ErrorMessage
	}
	now := time.Now()
	task.FinishedAt = &now

	if updateErr := s.compactBackupRepo.Save(ctx, task); updateErr != nil {
		logger.Error("[DOWNSTREAM_CHECK] failed to update compact backup status", zap.Error(updateErr))
		return updateErr
	}
	logger.Info("[DOWNSTREAM_CHECK] successfully updated local status")
	return nil
}

// handleDownstreamStatusResponse processes downstream status response and returns true if task is running
func (s *CompactBackupSchedulerImpl) handleDownstreamStatusResponse(
	ctx context.Context, task *entity.CompactBackup, resp *infraapi.CompactBackupResponse,
	taskLogger *zap.Logger,
) (bool, bizerr.BizError) {
	switch resp.Status {
	case infraapi.CompactBackupStatusComplete, infraapi.CompactBackupStatusFailed:
		if err := s.updateTaskStatus(ctx, task, resp, taskLogger); err != nil {
			return false, err
		}
		return false, nil

	case infraapi.CompactBackupStatusRunning, infraapi.CompactBackupStatusPending:
		taskLogger.Info("[DOWNSTREAM_CHECK] downstream task is still running",
			zap.String("status", string(resp.Status)))
		return true, nil

	case infraapi.CompactBackupStatusRetry:
		taskLogger.Info("[DOWNSTREAM_CHECK] downstream task is retrying",
			zap.String("status", string(resp.Status)))
		return true, nil

	default:
		taskLogger.Warn("[DOWNSTREAM_CHECK] unknown downstream compact backup status, considering as running",
			zap.String("status", string(resp.Status)))
		return true, nil
	}
}

// checkSingleTaskDownstreamStatus checks a single task's downstream status and returns true if it's running
func (s *CompactBackupSchedulerImpl) checkSingleTaskDownstreamStatus(
	ctx context.Context, task *entity.CompactBackup, cloudProvider, region string,
	logBackup *entity.LogBackup, logger *zap.Logger,
) (bool, bizerr.BizError) {
	taskLogger := logger.With(
		zap.Uint64("compact_backup_id", task.ID.Uint64()),
		zap.String("local_status", string(task.Status)),
		zap.String("downstream_id", *task.DownstreamCompactID),
	)

	taskLogger.Info("[DOWNSTREAM_CHECK] querying downstream compact backup status")

	// Query downstream status
	resp, facadeErr := s.compactFacade.GetCompactBackup(ctx, cloudProvider, region,
		logBackup.TenantID.Uint64(), logBackup.ProjectID.Uint64(), logBackup.ClusterID.Uint64(),
		*task.DownstreamCompactID)

	if facadeErr != nil {
		taskLogger.Error("[DOWNSTREAM_CHECK] failed to query downstream status",
			zap.Error(facadeErr), zap.String("error_type", fmt.Sprintf("%T", facadeErr)))

		if isNotFoundError(facadeErr) {
			// Downstream compact backup was deleted, update local status
			taskLogger.Info("[DOWNSTREAM_CHECK] downstream compact backup not found (deleted), updating local status to failed")

			task.Status = entity.CompactBackupStatusFailed
			task.ErrMsg = "Downstream compact backup was deleted"
			now := time.Now()
			task.FinishedAt = &now

			if updateErr := s.compactBackupRepo.Save(ctx, task); updateErr != nil {
				taskLogger.Error("[DOWNSTREAM_CHECK] failed to update compact backup status after downstream deletion",
					zap.Error(updateErr))
			}
			taskLogger.Info("[DOWNSTREAM_CHECK] successfully updated local status to failed")
			return false, nil
		}

		// Other query errors, conservatively consider task as potentially running
		taskLogger.Warn("[DOWNSTREAM_CHECK] other query error, considering task as running")
		return true, nil
	}

	taskLogger.Info("[DOWNSTREAM_CHECK] successfully queried downstream status",
		zap.String("downstream_status", string(resp.Status)))

	// Handle downstream status response
	return s.handleDownstreamStatusResponse(ctx, task, resp, taskLogger)
}

func (s *CompactBackupSchedulerImpl) hasRunningCompactTask(
	ctx context.Context,
	logBackup *entity.LogBackup,
) (bool, bizerr.BizError) {
	logger := log.GetLogger(ctx).With(
		zap.Uint64("log_backup_id", logBackup.ID.Uint64()),
		zap.Uint64("cluster_id", logBackup.ClusterID.Uint64()),
	)
	// Get all potentially running tasks from local DB
	allTasks, err := s.getLocalRunningTasks(ctx, logBackup.ID.Uint64())
	if err != nil {
		logger.Error("failed to query running/pending compact tasks", zap.Error(err))
		return false, err
	}

	if len(allTasks) == 0 {
		logger.Debug("no pending or running compact tasks found in local DB")
		return false, nil
	}

	logger.Debug("found potential running tasks in local DB", zap.Int("count", len(allTasks)))
	// Get cluster region
	cluster, bizErr := s.mgmtCli.GetDedicatedClusterBasicByID(ctx, logBackup.ClusterID.Uint64())
	if bizErr != nil {
		logger.Error("failed to fetch cluster information",
			zap.Uint64("cluster_id", logBackup.ClusterID.Uint64()), zap.Error(bizErr))
		return false, bizErr
	}
	cloudProvider := cluster.GetCloudProvider()
	region := cluster.GetRegion()

	logger.Debug("using region from cluster", zap.String("region", region))

	// Check each task's actual status from downstream
	var hasActuallyRunning bool
	for _, task := range allTasks {
		taskLogger := logger.With(zap.Uint64("compact_backup_id", task.ID.Uint64()))

		// Handle tasks without downstream ID
		if task.DownstreamCompactID == nil {
			if err := s.handleTaskWithoutDownstreamID(ctx, task, taskLogger); err != nil {
				logger.Error("failed to handle task without downstream ID", zap.Error(err))
			}
			continue
		}

		// Check downstream status for this task
		isRunning, checkErr := s.checkSingleTaskDownstreamStatus(ctx, task, cloudProvider, region, logBackup, taskLogger)
		if checkErr != nil {
			return false, checkErr
		}
		if isRunning {
			hasActuallyRunning = true
		}
	}

	logger.Info("checked running compact tasks", zap.Int("total_checked", len(allTasks)),
		zap.Bool("has_running", hasActuallyRunning))

	return hasActuallyRunning, nil
}

// isNotFoundError checks if the error indicates that the resource was not found
func isNotFoundError(err error) bool {
	if bizErr, ok := err.(bizerr.BizError); ok {
		isNotFound := bizErr.ErrCode() == bizerr.ErrCodeRecordNotFound
		log.GetLogger(context.Background()).Debug("[ERROR_CHECK] checking if error is not found",
			zap.Error(err),
			zap.Int64("error_code", int64(bizErr.ErrCode())),
			zap.Bool("is_not_found", isNotFound))
		return isNotFound
	}
	log.GetLogger(context.Background()).Debug("[ERROR_CHECK] error is not BizError",
		zap.Error(err),
		zap.String("error_type", fmt.Sprintf("%T", err)))
	return false
}

func (s *CompactBackupSchedulerImpl) createCompactTask(
	ctx context.Context,
	logger *zap.Logger,
	logBackup *entity.LogBackup,
) bizerr.BizError {
	req := &domain.AsyncCreateCompactBackupReq{
		ClusterID:   logBackup.ClusterID.Uint64(),
		LogBackupID: logBackup.ID.Uint64(),
		UserID:      0, // TODO: what is this user ID?
		ProjectID:   0, // TODO: what is this project ID?
		IsInternal:  true,
	}

	resp, err := s.compactBackupModel.AsyncCreate(ctx, req)
	if err != nil {
		logger.Error("failed to create compact backup", zap.Error(err))
		return err
	}

	logger.Info("compact backup created successfully",
		zap.Uint64("compact_backup_id", resp.CompactBackupID),
		zap.String("workflow_id", resp.WorkflowID),
		zap.String("start_ts", *logBackup.StartTS),
		zap.String("end_ts", logBackup.CheckpointTS))

	return nil
}