package console

import (
	"context"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/elliotchance/pie/v2"
	"github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/restore"
	backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/pkg/account"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/tiinf/sdk/auditlog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/timestamppb"
	"gorm.io/gorm"
)

type BackupControllerV2 interface {
	dataflow.BackupConsoleServiceServer
}

type backupControllerV2Impl struct {
	dataflow.UnimplementedBackupConsoleServiceServer
	*BaseController

	restoreRouter       restore.RestoreModel
	snapshotBackupModel backup.SnapshotBackupModel
	backupSettingModel  backup.BackupSettingModel
	backupCreateRouter  backup.BackupCreateRouter
	backupExportRepo    backupRepo.BackupExportRepo
	backupExportModel   backup.BackupExportModel
	backupRepo          backupRepo.BackupRepo
}

func NewBackupControllerV2(
	trans db.TransProvider,
	restoreRouter restore.RestoreModel,
	snapshotBackupModel backup.SnapshotBackupModel,
	backupSettingModel backup.BackupSettingModel,
	backupCreateRouter backup.BackupCreateRouter,
	backupExportRepo backupRepo.BackupExportRepo,
	backupExportModel backup.BackupExportModel,
	backupRepo backupRepo.BackupRepo,
) BackupControllerV2 {
	return &backupControllerV2Impl{
		BaseController:      NewBaseController(trans),
		snapshotBackupModel: snapshotBackupModel,
		backupSettingModel:  backupSettingModel,
		backupCreateRouter:  backupCreateRouter,
		restoreRouter:       restoreRouter,
		backupExportRepo:    backupExportRepo,
		backupExportModel:   backupExportModel,
		backupRepo:          backupRepo,
	}
}

func (c *backupControllerV2Impl) GetBackupExportProfile(
	ctx context.Context, req *dataflow.GetBackupExportProfileReq,
) (*dataflow.GetBackupExportProfileResp, error) {
	_, err := c.checkGetBackupExportProfileReq(ctx, req)
	if err != nil {
		return nil, err
	}

	profile, err := c.backupExportModel.GetBackupExportProfile(ctx, &domain.GetBackupExportProfileReq{
		BackupID: req.BackupId,
	})
	if err != nil {
		return nil, err
	}
	resp := &dataflow.GetBackupExportProfileResp{}
	resp.GcpServiceAccount = profile.GCPServiceAccount
	resp.S3RoleArn = profile.S3RoleArn
	resp.AzBlobRoleArn = profile.AzBlobRoleArn

	return resp, nil
}

func (c *backupControllerV2Impl) CreateBackupExport(
	ctx context.Context, req *dataflow.CreateBackupExportReq,
) (resp *dataflow.CreateBackupExportResp, err error) {
	resp = &dataflow.CreateBackupExportResp{}
	defer func() {
		c.recordAuditLogForCreateBackupExport(ctx, req, resp, err)
	}()
	ctx = log.WithTraceID(ctx)

	_, err = c.checkCreateBackupExportReqIsValid(ctx, req)
	if err != nil {
		return nil, err
	}

	user := account.GetUserFromContext(ctx)
	err = c.trans.Transaction(ctx, func(ctx context.Context) error {
		exportId, err := c.backupExportModel.CreateBackupExport(ctx, req, user)
		if err != nil {
			return err
		}
		resp.ExportId = exportId
		return nil
	})
	if err != nil {
		return nil, err
	}

	return resp, nil
}

func (c *backupControllerV2Impl) CancelBackupExport(
	ctx context.Context, req *dataflow.CancelBackupExportReq,
) (*dataflow.CancelBackupExportResp, error) {
	ctx = log.WithTraceID(ctx)

	_, err := c.checkCancelBackupExportReqIsValid(ctx, req)
	if err != nil {
		return nil, err
	}

	var resp *dataflow.CancelBackupExportResp

	if err := c.trans.Transaction(ctx, func(ctx context.Context) error {
		err := c.backupExportModel.CancelBackupExport(ctx, &dataflow.CancelBackupExportReq{
			ExportId: req.ExportId,
		})
		if err != nil {
			return err
		}
		return nil
	}); err != nil {
		return nil, err
	}

	return resp, nil
}

func (c *backupControllerV2Impl) ListBackupExports(
	ctx context.Context, req *dataflow.ListBackupExportsReq,
) (*dataflow.ListBackupExportsResp, error) {
	ctx = log.WithTraceID(ctx)

	filter := &backupRepo.BackupExportFilter{
		TenantID:  pointer.To(req.OrgId),
		ProjectID: pointer.To(req.ProjectId),
		ClusterID: pointer.To(req.ClusterId),
	}
	// Get total count
	count, err := c.backupExportRepo.CountByFilter(ctx, filter)
	if err != nil {
		return nil, err
	}

	// Get list of backup exports
	backupExports, err := c.backupExportRepo.ListByFilter(
		ctx,
		filter,
		func(tx *gorm.DB) *gorm.DB {
			pager := domain.NewPager(domain.PageSizeOp(int64(req.PerPage)), domain.PageNumberOp(int64(req.Page)))
			if pager == nil {
				pager = domain.NewPager()
			}
			offset, limit := pager.CalculateOffsetAndLimit()
			return tx.Offset(int(offset)).Limit(int(limit))
		},
		func(tx *gorm.DB) *gorm.DB {
			return tx.Order("created_at DESC")
		},
	)
	if err != nil {
		return nil, err
	}

	// Convert to response format
	items := make([]*dataflow.BackupExport, 0, len(backupExports))
	for _, export := range backupExports {
		items = append(items, domain.BackupExportEntity2Dataflow(ctx, &export))
	}

	return &dataflow.ListBackupExportsResp{
		Total: uint64(count),
		Items: items,
	}, nil
}

func (c *backupControllerV2Impl) QueryBackupSetting(
	ctx context.Context, req *dataflow.QueryBackupSettingReq,
) (*dataflow.QueryBackupSettingResp, error) {
	ctx = log.WithTraceID(ctx)
	queryResp, err := c.backupSettingModel.QueryBackupSettingForAPI(ctx, pointer.ToOrNil(req.ClusterId))
	if err != nil {
		if err.ErrCode() == config.ErrCodeBackupSettingIsNotReady {
			return &dataflow.QueryBackupSettingResp{}, nil
		}
		return nil, err
	}

	resp, err := c.convertQueryBackupSettingResp(ctx, queryResp)
	if err != nil {
		return nil, err
	}
	return resp, nil
}

func (c *backupControllerV2Impl) convertQueryBackupSettingResp(
	ctx context.Context, queryResp *domain.QueryBackupSettingForAPIResp,
) (*dataflow.QueryBackupSettingResp, bizerr.BizError) {
	resp := &dataflow.QueryBackupSettingResp{}
	status, actions := domain.ConvertBackupSettingStatusMatrixToDataflow(
		queryResp.Status, queryResp.AllowActions, queryResp.PitBackupSwitch.Status)
	snapshotBackupSchedule := c.convertSnapshotBackupScheduleToResp(ctx, queryResp)
	continuousBackupSchedule := c.convertContinuousBackupScheduleToResp(queryResp)
	regions := c.convertBackupSettingRegionsToResp(queryResp)

	resp.Status = status
	resp.Actions = actions
	resp.SnapshotBackupSchedule = snapshotBackupSchedule
	resp.ContinuousBackupSchedule = continuousBackupSchedule
	resp.Regions = regions
	resp.CrrfStatus = entityStatusToCRRFProtoStatus(queryResp.CRRFStatus)
	return resp, nil
}

func (c *backupControllerV2Impl) convertBackupSettingRegionsToResp(
	queryResp *domain.QueryBackupSettingForAPIResp,
) []*dataflow.RegionInfo {
	if queryResp.BucketSetting == nil {
		return nil
	}
	var regionInfos []*dataflow.RegionInfo
	pie.Of(queryResp.BucketSetting.Buckets).Each(func(bucket domain.BackupSettingBucket) {
		pie.Of(bucket.Regions).Each(func(region domain.RegionInfo) {
			regionInfos = append(regionInfos, &dataflow.RegionInfo{
				Name:        region.Name,
				DisplayName: region.DisplayName,
			})
		})
	})
	return regionInfos
}

func (c *backupControllerV2Impl) convertContinuousBackupScheduleToResp(
	queryResp *domain.QueryBackupSettingForAPIResp,
) *dataflow.ContinuousBackupSchedule {
	if queryResp.PiTRSetting == nil {
		return nil
	}
	continuousBackupSchedule := &dataflow.ContinuousBackupSchedule{
		Enable: queryResp.PiTRSetting.Enable,
	}
	return continuousBackupSchedule
}

func (c *backupControllerV2Impl) convertSnapshotBackupScheduleToResp(
	ctx context.Context, queryResp *domain.QueryBackupSettingForAPIResp,
) *dataflow.SnapshotBackupSchedule {
	if queryResp.SnapshotBackupSchedule == nil {
		return nil
	}
	snapshotBackupSchedule := &dataflow.SnapshotBackupSchedule{
		Disable:           queryResp.SnapshotBackupSchedule.Disable,
		ScheduleType:      domain.SnapshotBackupScheduleDomain2Dataflow(queryResp.SnapshotBackupSchedule.ScheduleType),
		DailyScheduleTime: queryResp.SnapshotBackupSchedule.DailyScheduleTime,
	}
	if queryResp.SnapshotBackupSchedule.WeeklySchedule != nil {
		snapshotBackupSchedule.WeeklySchedule = &dataflow.WeeklySchedule{
			Days:         domain.WeekdayDomainList2Dataflow(queryResp.SnapshotBackupSchedule.WeeklySchedule.Days),
			ScheduleTime: queryResp.SnapshotBackupSchedule.WeeklySchedule.ScheduleTime,
		}
	}
	if queryResp.SnapshotBackupSchedule.RetentionPolicy != nil {
		snapshotBackupSchedule.RetentionPolicy = &dataflow.RetentionPolicy{
			RetentionPeriod: int64(queryResp.SnapshotBackupSchedule.RetentionPolicy.RetentionPeriod),
		}
	}
	if queryResp.SnapshotBackupSchedule.NextBackupTime != nil {
		nextBackupTime := timestamppb.New(*queryResp.SnapshotBackupSchedule.NextBackupTime)
		if err := nextBackupTime.CheckValid(); err != nil {
			log.GetLogger(ctx).Warn("failed to parse next backup time", zap.Error(err))
		} else {
			snapshotBackupSchedule.NextBackupTime = nextBackupTime
		}
	}
	return snapshotBackupSchedule
}

func (c *backupControllerV2Impl) ModifyBackupSetting(
	ctx context.Context, req *dataflow.ModifyBackupSettingReqV2,
) (_ *dataflow.ModifyBackupSettingRespV2, err error) {
	ctx = log.WithTraceID(ctx)
	var beforeSetting *domain.QueryBackupSettingForAPIResp
	defer func() {
		c.recordAuditLogForModifyBackupSetting(ctx, req, beforeSetting, err)
	}()
	setting, err := c.backupSettingModel.QueryBackupSettingForAPI(ctx, pointer.ToOrNil(req.ClusterId))
	if err != nil {
		return nil, err
	}
	beforeSetting = setting
	err = c.checkModifyBackupSettingParams(ctx, req)
	if err != nil {
		return nil, err
	}

	changedInfo, err := c.calculateBackupSettingChangedInfo(ctx, req, setting)
	if err != nil {
		return nil, err
	}

	if changedInfo.piTRSettingChanged {
		if err := c.modifyContinuousBackupScheduleSetting(ctx, req); err != nil {
			return nil, err
		}
		if err := c.modifyAutoBackupSetting(ctx, req); err != nil {
			return nil, err
		}
		return &dataflow.ModifyBackupSettingRespV2{}, nil
	}

	if changedInfo.dualRegionSettingChanged {
		if err := c.modifyDualRegionBackupSetting(ctx, req, changedInfo); err != nil {
			return nil, err
		}
		if err := c.modifyAutoBackupSetting(ctx, req); err != nil {
			return nil, err
		}
		return &dataflow.ModifyBackupSettingRespV2{}, nil
	}

	if err := c.modifyAutoBackupSetting(ctx, req); err != nil {
		return nil, err
	}
	return &dataflow.ModifyBackupSettingRespV2{}, nil
}

func (c *backupControllerV2Impl) recordAuditLogForModifyBackupSetting(
	ctx context.Context,
	req *dataflow.ModifyBackupSettingReqV2,
	beforeSetting *domain.QueryBackupSettingForAPIResp,
	rErr error,
) {
	logger := log.GetLogger(ctx)
	if req.DryRun {
		return
	}

	logDetail := &ModifyBackupSettingAuditLogDetail{
		Error: c.ConvertAuditLogDetailErrMsg(rErr),
	}
	logDetail.Before = c.convertSettingToAuditLogDetailItem(beforeSetting)
	if rErr == nil {
		afterSetting, err := c.backupSettingModel.QueryBackupSettingForAPI(ctx, pointer.ToOrNil(req.ClusterId))
		if err != nil {
			logger.Warn("audit log failed, failed to query backup setting after modify", zap.Error(err))
			return
		}
		logDetail.After = c.convertSettingToAuditLogDetailItem(afterSetting)
	}

	details, err := c.ConvertToAuditLogDetail(logDetail)
	if err != nil {
		logger.Warn("audit log failed, failed to convert to audit log detail", zap.Error(err))
		return
	}

	err = auditlog.NewEvent(ctx, auditlog.EventSetAutoBackup).
		SetResult(rErr == nil).
		SetOrgID(req.OrgId).
		SetProjectID(req.ProjectId).
		SetDetails(details.Build()).
		Submit()
	if err != nil {
		log.GetLogger(ctx).Warn("failed to submit audit log", zap.Error(err))
		return
	}
	return
}

func (c *backupControllerV2Impl) convertSettingToAuditLogDetailItem(
	setting *domain.QueryBackupSettingForAPIResp,
) *ModifyBackupSettingAuditLogDetailItem {
	if setting == nil {
		return nil
	}
	detailItem := &ModifyBackupSettingAuditLogDetailItem{}
	detailItem.SnapshotBackupSchedule = setting.SnapshotBackupSchedule
	if detailItem.SnapshotBackupSchedule != nil {
		detailItem.SnapshotBackupSchedule.NextBackupTime = nil
	}
	detailItem.PiTR = setting.PiTRSetting
	if setting.BucketSetting != nil {
		pie.Of(setting.BucketSetting.Buckets).Each(func(bucket domain.BackupSettingBucket) {
			pie.Of(bucket.Regions).Each(func(region domain.RegionInfo) {
				detailItem.Regions = append(detailItem.Regions, region.DisplayName)
			})
		})
	}
	return detailItem
}

func (c *backupControllerV2Impl) modifyAutoBackupSetting(
	ctx context.Context, req *dataflow.ModifyBackupSettingReqV2,
) bizerr.BizError {
	modifyAutoBackupReq := &domain.ModifyAutoBackupSettingReq{
		ClusterID: req.ClusterId,
		SnapshotBackupSchedule: domain.SnapshotBackupSchedule{
			Disable:      req.SnapshotBackupSchedule.Disable,
			ScheduleType: domain.SnapshotBackupScheduleDataflow2Domain(req.SnapshotBackupSchedule.ScheduleType),
			RetentionPolicy: &domain.RetentionPolicy{
				RetentionPeriod: int(req.SnapshotBackupSchedule.RetentionPolicy.RetentionPeriod),
			},
		},
		DryRun: req.DryRun,
	}
	if req.SnapshotBackupSchedule.ScheduleType == dataflow.SnapshotBackupScheduleType_Daily {
		modifyAutoBackupReq.SnapshotBackupSchedule.DailyScheduleTime = req.SnapshotBackupSchedule.DailyScheduleTime
	} else {
		modifyAutoBackupReq.SnapshotBackupSchedule.WeeklySchedule = &domain.WeeklySchedule{
			Days:         domain.WeekdayDataflowList2Domain(req.SnapshotBackupSchedule.WeeklySchedule.Days),
			ScheduleTime: req.SnapshotBackupSchedule.WeeklySchedule.ScheduleTime,
		}
	}

	log.GetLogger(ctx).Info("modify auto backup setting", zap.Any("req", modifyAutoBackupReq))
	if _, err := c.backupSettingModel.ModifyAutoBackupSetting(ctx, modifyAutoBackupReq); err != nil {
		return err
	}
	return nil
}

func (c *backupControllerV2Impl) modifyDualRegionBackupSetting(
	ctx context.Context, req *dataflow.ModifyBackupSettingReqV2, changedInfo *backupSettingChangeInfo,
) bizerr.BizError {
	regionNames := pie.Map(req.Regions, func(region *dataflow.RegionInfo) string {
		return region.Name
	})
	appReq := &domain.RegionSettingModificationReqV2{
		ClusterID: &req.ClusterId,
		Enable:    changedInfo.requestDualRegionEnable,
		Regions:   regionNames,
		DryRun:    req.DryRun,
	}
	if _, err := c.backupSettingModel.ModifyRegionSetting(ctx, appReq); err != nil {
		return err
	}
	log.GetLogger(ctx).Info("modify backup region setting success")
	return nil
}

func (c *backupControllerV2Impl) modifyContinuousBackupScheduleSetting(
	ctx context.Context, req *dataflow.ModifyBackupSettingReqV2,
) bizerr.BizError {
	appReq := &domain.PitSettingModificationReq{
		ClusterID: req.ClusterId,
		Enable:    req.ContinuousBackupSchedule.Enable,
		DryRun:    req.DryRun,
	}
	if err := c.backupSettingModel.ModifyPitBackupSetting(ctx, appReq); err != nil {
		return err
	}
	log.GetLogger(ctx).Info("modify pitr success")
	return nil
}

type backupSettingChangeInfo struct {
	dualRegionSettingChanged bool
	requestDualRegionEnable  bool
	currentDualRegionEnable  bool

	piTRSettingChanged bool
	requestPiTREnable  bool
	currentPiTREnable  bool

	closePiTRIfSnapshotBackupScheduleRequestClosed bool
}

func (c *backupControllerV2Impl) calculateBackupSettingChangedInfo(
	ctx context.Context, req *dataflow.ModifyBackupSettingReqV2, setting *domain.QueryBackupSettingForAPIResp,
) (*backupSettingChangeInfo, bizerr.BizError) {
	logger := log.GetLogger(ctx)

	info := &backupSettingChangeInfo{
		closePiTRIfSnapshotBackupScheduleRequestClosed: c.closePiTRIfSnapshotBackupScheduleRequestClosed(ctx, req),
	}

	c.calculatePiTRChangedInfo(info, setting, req)
	c.calculateDualRegionChangedInfo(info, setting, req)

	if info.dualRegionSettingChanged && info.piTRSettingChanged {
		if info.closePiTRIfSnapshotBackupScheduleRequestClosed {
			logger.Warn("dual region and pitr setting change simultaneously, beacause of closing schedule")
			return nil, bizerr.New(
				config.ErrCodeBackupSettingChangeSimultaneously,
				config.ErrMsgBackupSettingChangeSimultaneouslyDueToDisablingSchedule,
			)
		}
		logger.Warn("dual region and pitr setting change simultaneously")
		return nil, bizerr.New(
			config.ErrCodeBackupSettingChangeSimultaneously,
			config.ErrMsgBackupSettingChangeSimultaneously,
		)
	}

	return info, nil
}

func (c *backupControllerV2Impl) calculateDualRegionChangedInfo(
	info *backupSettingChangeInfo, setting *domain.QueryBackupSettingForAPIResp, req *dataflow.ModifyBackupSettingReqV2,
) {
	info.currentDualRegionEnable = setting.BucketSetting.DualRegionEnable
	if len(req.Regions) == 0 {
		info.dualRegionSettingChanged = false
		info.requestDualRegionEnable = info.currentDualRegionEnable
	} else {
		info.requestDualRegionEnable = len(req.Regions) > 1
		info.dualRegionSettingChanged = info.requestDualRegionEnable != info.currentDualRegionEnable
	}
}

func (c *backupControllerV2Impl) calculatePiTRChangedInfo(
	info *backupSettingChangeInfo, setting *domain.QueryBackupSettingForAPIResp, req *dataflow.ModifyBackupSettingReqV2,
) {
	info.currentPiTREnable = setting.PiTRSetting.Enable
	if req.ContinuousBackupSchedule == nil {
		info.piTRSettingChanged = false
		info.requestPiTREnable = info.currentPiTREnable
	} else {
		info.requestPiTREnable = req.ContinuousBackupSchedule != nil && req.ContinuousBackupSchedule.Enable
		info.piTRSettingChanged = info.requestPiTREnable != info.currentPiTREnable
	}
}

func (c *backupControllerV2Impl) closePiTRIfSnapshotBackupScheduleRequestClosed(
	ctx context.Context, req *dataflow.ModifyBackupSettingReqV2,
) bool {
	if req.SnapshotBackupSchedule == nil || !req.SnapshotBackupSchedule.Disable {
		return false
	}
	if req.ContinuousBackupSchedule == nil {
		req.ContinuousBackupSchedule = &dataflow.ContinuousBackupSchedule{
			Enable: false,
		}
		return true
	}
	if req.ContinuousBackupSchedule.Enable {
		req.ContinuousBackupSchedule.Enable = false
		return true
	}
	return false
}

func (c *backupControllerV2Impl) checkModifyBackupSettingParams(
	ctx context.Context, req *dataflow.ModifyBackupSettingReqV2,
) error {
	logger := log.GetLogger(ctx)
	if req.SnapshotBackupSchedule == nil || req.ContinuousBackupSchedule == nil || len(req.Regions) == 0 {
		logger.Warn(config.ErrMsgModifyBackupSettingInputIsEmpty)
		return bizerr_wrapper.NewInvalidInputError(config.ErrMsgModifyBackupSettingInputIsEmpty)
	}
	return c.checkModifySnapshotBackupScheduleParams(ctx, req)
}

func (c *backupControllerV2Impl) checkModifySnapshotBackupScheduleParams(
	ctx context.Context, req *dataflow.ModifyBackupSettingReqV2,
) bizerr.BizError {
	if req.SnapshotBackupSchedule == nil {
		return nil
	}
	logger := log.GetLogger(ctx)

	if err := utils.CheckParamEmptyError(
		"retention policy", req.SnapshotBackupSchedule.RetentionPolicy); err != nil {
		logger.Warn("invalid param", zap.Error(err))
		return err
	}
	if err := utils.CheckParamEmptyError(
		"retention period", req.SnapshotBackupSchedule.RetentionPolicy.RetentionPeriod); err != nil {
		logger.Warn("invalid param", zap.Error(err))
		return err
	}
	if req.SnapshotBackupSchedule.ScheduleType == dataflow.SnapshotBackupScheduleType_Weekly {
		if err := utils.CheckParamEmptyError(
			"weekly schedule", req.SnapshotBackupSchedule.WeeklySchedule); err != nil {
			logger.Warn("invalid param", zap.Error(err))
			return err
		}
		if err := utils.CheckParamEmptyError(
			"weekly schedule time", req.SnapshotBackupSchedule.WeeklySchedule.ScheduleTime); err != nil {
			logger.Warn("invalid param", zap.Error(err))
			return err
		}
		if err := utils.CheckParamEmptyError(
			"weekly schedule days", req.SnapshotBackupSchedule.WeeklySchedule.Days); err != nil {
			logger.Warn("invalid param", zap.Error(err))
			return err
		}
	}
	if req.SnapshotBackupSchedule.ScheduleType == dataflow.SnapshotBackupScheduleType_Daily {
		if err := utils.CheckParamEmptyError(
			"daily schedule time", req.SnapshotBackupSchedule.DailyScheduleTime); err != nil {
			logger.Warn("invalid param", zap.Error(err))
			return err
		}
	}
	return nil
}

func (c *backupControllerV2Impl) ListRestores(
	ctx context.Context, req *dataflow.ListRestoresReq,
) (*dataflow.ListRestoresResp, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)

	listRestoreReq := domain.ListRestoreReqV2{
		Filter: &domain.ListRestoreFilter{
			TenantIDs:        []uint64{req.OrgId},
			ProjectIDs:       []uint64{req.ProjectId},
			BackupClusterIDs: []uint64{req.ClusterId},
		},
	}
	listRestoreReq.Pager = domain.NewPager(domain.PageSizeOp(int64(req.PerPage)), domain.PageNumberOp(int64(req.Page)))
	restoreRespV2, err := c.restoreRouter.List(ctx, &listRestoreReq)
	if err != nil {
		return nil, err
	}

	items := pie.Map(restoreRespV2.Items, func(item domain.ListRestoreRespV2Item) *dataflow.ListRestoresItem {
		respItem := &dataflow.ListRestoresItem{
			Id:            item.Id,
			ClusterId:     item.RestoreClusterID,
			Name:          item.RestoreClusterName,
			RestoreMode:   domain.RestoreModeDomain2Dataflow(item.Mode),
			Status:        domain.RestoreStatusDomain2Dataflow(item.Status),
			Progress:      item.Progress,
			RestoreRegion: item.RestoreClusterRegion,
			CreatedAt:     timestamppb.New(item.CreateTime),
			RestoredTso:   item.RestoreTSO,
			BackupName:    item.BackupName,
		}
		if item.RestoreTS != nil {
			restoreTs := timestamppb.New(time.Unix(*item.RestoreTS, 0))
			if err := restoreTs.CheckValid(); err != nil {
				logger.Warn("failed to parse restoreTs",
					zap.Uint64("clusterID", item.RestoreClusterID), zap.Error(err))
			} else {
				respItem.RestoredTs = restoreTs
			}
		}
		if item.FinishedTime != nil {
			respItem.FinishedAt = timestamppb.New(*item.FinishedTime)
		}
		return respItem
	})
	return &dataflow.ListRestoresResp{
		Items: items,
		Total: uint32(restoreRespV2.Total),
	}, nil
}

func (c *backupControllerV2Impl) checkGetBackupExportProfileReq(
	ctx context.Context, req *dataflow.GetBackupExportProfileReq,
) (*entity.Backup, bizerr.BizError) {
	backup, err := c.backupRepo.Load(ctx, req.BackupId)
	if err != nil {
		return nil, err
	}
	if backup.ProjectID != entity.ID(req.ProjectId) || backup.TenantID != entity.ID(req.OrgId) {
		return nil, bizerr_wrapper.NewInvalidInputError("invalid project_id or org_id")
	}
	return backup, nil
}

func (c *backupControllerV2Impl) checkCreateBackupExportReqIsValid(
	ctx context.Context, req *dataflow.CreateBackupExportReq,
) (*entity.Backup, bizerr.BizError) {
	backup, err := c.backupRepo.Load(ctx, req.BackupId)
	if err != nil {
		return nil, err
	}
	if backup.ProjectID != entity.ID(req.ProjectId) || backup.TenantID != entity.ID(req.OrgId) {
		return nil, bizerr_wrapper.NewInvalidInputError("invalid project_id or org_id")
	}

	if backup.Status != entity.BackupStatusSuccess {
		return nil, bizerr_wrapper.NewInvalidInputError("backup is not completed")
	}
	if *backup.CleanedUp {
		return nil, bizerr_wrapper.NewInvalidInputError("backup is already cleaned up")
	}
	return backup, nil
}

func (c *backupControllerV2Impl) checkCancelBackupExportReqIsValid(
	ctx context.Context, req *dataflow.CancelBackupExportReq,
) (*entity.BackupExport, bizerr.BizError) {
	export, err := c.backupExportRepo.Load(ctx, req.ExportId)
	if err != nil {
		return nil, err
	}
	if export.ProjectID != entity.ID(req.ProjectId) ||
		export.TenantID != entity.ID(req.OrgId) ||
		export.ClusterID != entity.ID(req.ClusterId) ||
		export.BackupID != entity.ID(req.BackupId) {
		return nil, bizerr_wrapper.NewInvalidInputError("invalid project_id or org_id or cluster_id or backup_id")
	}

	if !export.IsCancellable() {
		return nil, bizerr_wrapper.NewInvalidInputError("backup export is not cancellable, current status: %s", export.Status)
	}

	return export, nil
}

func (c *backupControllerV2Impl) recordAuditLogForCreateBackupExport(
	ctx context.Context, req *dataflow.CreateBackupExportReq, resp *dataflow.CreateBackupExportResp, rErr error,
) {
	return
	/*
		logger := log.GetLogger(ctx)
		backup, err := c.backupRepo.Load(ctx, req.BackupId)
		if err != nil || backup == nil {
			logger.Warn("failed to submit audit log, query backup error",
				zap.Uint64("backupID", req.BackupId), zap.Error(err))
			return
		}
		logDetail := &CreateBackupExportAuditLogDetail{
			BackupID:         req.BackupId,
			BackupName:       backup.BackupSchedule.Name,
			BackupExportID:   resp.ExportId,
			BackupExportName: req.BackupExport.Name,
			Error:            c.ConvertAuditLogDetailErrMsg(rErr),
		}

		backupExport, err := c.backupExportRepo.Load(ctx, resp.ExportId)
		if err != nil || backupExport == nil {
			logger.Warn("failed to submit audit log, query backup error",
				zap.Uint64("backupID", req.Id), zap.Error(err))
			return
		}

		details, ierr := c.ConvertToAuditLogDetail(logDetail)
		if ierr != nil {
			logger.Warn("failed to convert to audit log detail", zap.Error(ierr))
			return
		}
		cluster, bizError := c.coreCli.GetAnyClusterByID(ctx, req.ClusterId)
		if bizError != nil {
			logger.Warn("failed to submit audit log, query cluster error",
				zap.Uint64("clusterId", req.ClusterId), zap.Error(bizError))
			return
		}
		eventType := auditlog.EventRecoverFromBackup
		if grpcli.IsClusterDeleted(cluster.Status) {
			eventType = auditlog.EventRecoverFromTrash
		}

		event := auditlog.NewEvent(ctx, eventType).
			SetResult(rErr == nil).
			SetOrgID(req.OrgId).
			SetProjectID(req.ProjectId).
			SetDetails(details.Build())
		user := account.GetUserFromContext(ctx)
		if user.Email == KeyOpenAPI {
			event.SetAPIPubKey(user.LastName)
		}
		if err := event.Submit(); err != nil {
			logger.Warn("failed to submit audit log", zap.Error(err))
		}
		return
	*/
}