package console

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

	"github.com/AlekSi/pointer"
	"github.com/Masterminds/semver/v3"
	"github.com/elliotchance/pie/v2"
	"github.com/juju/errors"
	"github.com/tidbcloud/dataflow-service/app/config"
	"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"
	"github.com/tidbcloud/dataflow-service/app/models/restore"
	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"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	internal_config "github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/model/util"
	"github.com/tidbcloud/dataflow-service/pkg/account"
	"github.com/tidbcloud/dataflow-service/pkg/billing"
	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"
	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"
	"github.com/tidbcloud/tiinf/sdk/auditlog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

const KeyOpenAPI = "OpenAPI"

type BackupController interface {
	dataflow.BackupServiceServer
}

type backupControllerImpl struct {
	dataflow.UnimplementedBackupServiceServer
	*BaseController

	backupRepo          backup_repo.BackupRepo
	restoreRouter       restore.RestoreModel
	snapshotBackupModel backup.SnapshotBackupModel
	backupSettingModel  backup.BackupSettingModel
	backupCreateRouter  backup.BackupCreateRouter
	PitBackupModel      backup.PitBackupModel

	coreCli grpcli.CoreGrpcClient
	mgmtCli grpcli.MgmtSvcGrpcClient

	billingHandler billing.BillingHandler

	featureGate feature_gate.FeatureGate
}

func NewBackupController(
	trans db.TransProvider,
	restoreRouter restore.RestoreModel,
	snapshotBackupModel backup.SnapshotBackupModel,
	backupSettingModel backup.BackupSettingModel,
	backupCreateRouter backup.BackupCreateRouter,
	pitBackupModel backup.PitBackupModel,
	backupRepo backup_repo.BackupRepo,
	coreCli grpcli.CoreGrpcClient,
	mgmtCli grpcli.MgmtSvcGrpcClient,
	billingHandler billing.BillingHandler,
	featureGate feature_gate.FeatureGate,
) BackupController {
	return &backupControllerImpl{
		BaseController:      NewBaseController(trans),
		restoreRouter:       restoreRouter,
		snapshotBackupModel: snapshotBackupModel,
		backupSettingModel:  backupSettingModel,
		backupCreateRouter:  backupCreateRouter,
		PitBackupModel:      pitBackupModel,
		backupRepo:          backupRepo,
		coreCli:             coreCli,
		mgmtCli:             mgmtCli,
		billingHandler:      billingHandler,
		featureGate:         featureGate,
	}
}

func (c *backupControllerImpl) Restore(
	ctx context.Context, req *dataflow.BackupRestoreReq,
) (_ *dataflow.RestoreResp, rErr error) {
	defer func() {
		c.recordAuditLogForRestore(ctx, req, rErr)
	}()
	ctx = log.WithTraceID(ctx)

	if req.RestoreMode == dataflow.RestoreMode_RESTORE_MODE_SNAPSHOT {
		_, err := c.checkBackupPermission(ctx, req.OrgId, req.ProjectId, req.ClusterId, req.Id)
		if err != nil {
			return nil, err
		}
	} else if req.RestoreMode == dataflow.RestoreMode_RESTORE_MODE_PITR {
		// PiTR restore does not provide backup_id, for example:
		// `/api/v1/orgs/{org_id}/projects/{project_id}/clusters/{cluster_id}/backups/0/restore`
		_, err := c.checkClusterPermission(ctx, req.OrgId, req.ProjectId, req.ClusterId)
		if err != nil {
			return nil, err
		}
	}

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

	restoreReq := c.buildCreateRestoreReq(ctx, req)
	createRestoreResp, err := c.restoreRouter.Create(ctx, restoreReq)
	if err != nil {
		return nil, err
	}
	return &dataflow.RestoreResp{
		Id:        createRestoreResp.ID,
		ClusterId: createRestoreResp.ClusterID,
		OrgId:     req.OrgId,
	}, nil
}

func (c *backupControllerImpl) recordAuditLogForRestore(
	ctx context.Context, restoreReq *dataflow.BackupRestoreReq, rErr error,
) {
	if restoreReq.RestoreMode == dataflow.RestoreMode_RESTORE_MODE_PITR {
		c.recordAuditLogForPiTRRestore(ctx, restoreReq, rErr)
		return
	}

	c.recordAuditLogForSnapshotRestore(ctx, restoreReq, rErr)
}

func (c *backupControllerImpl) recordAuditLogForPiTRRestore(
	ctx context.Context, restoreReq *dataflow.BackupRestoreReq, rErr error,
) {
	logger := log.GetLogger(ctx)

	logDetail := &PitRestoreAuditLogDetail{
		BackupClusterID: restoreReq.ClusterId,
		RestoredTs:      restoreReq.RestoredTs,
		RestoreCluster:  c.buildClusterConfig(restoreReq),
		Error:           c.ConvertAuditLogDetailErrMsg(rErr),
	}
	details, err := c.ConvertToAuditLogDetail(logDetail)
	if err != nil {
		logger.Warn("failed to convert to audit log detail", zap.Error(err))
		return
	}
	event := auditlog.NewEvent(ctx, auditlog.EventRestoreFromPITR).
		SetResult(rErr == nil).
		SetOrgID(restoreReq.OrgId).
		SetProjectID(restoreReq.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
}

func (c *backupControllerImpl) recordAuditLogForSnapshotRestore(
	ctx context.Context, restoreReq *dataflow.BackupRestoreReq, rErr error,
) {
	logger := log.GetLogger(ctx)
	resp, err := c.snapshotBackupModel.Query(ctx, &domain.QueryBackupReq{
		Filter: &domain.ListBackupFilter{
			BackupIDs: []uint64{restoreReq.Id},
		},
	})
	if err != nil || resp.BackupEntity == nil || resp.BackupEntity.BackupSchedule == nil {
		logger.Warn("failed to submit audit log, query backup error",
			zap.Uint64("backupID", restoreReq.Id), zap.Error(err))
		return
	}

	logDetail := &SnapshotRestoreAuditLogDetail{
		BackupClusterID: restoreReq.ClusterId,
		BackupID:        restoreReq.Id,
		BackupName:      resp.BackupEntity.BackupSchedule.Name,
		RestoreCluster:  c.buildClusterConfig(restoreReq),
		Error:           c.ConvertAuditLogDetailErrMsg(rErr),
	}

	details, ierr := c.ConvertToAuditLogDetail(logDetail)
	if ierr != nil {
		logger.Warn("failed to convert to audit log detail", zap.Error(ierr))
		return
	}
	cluster, bizError := c.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, restoreReq.ClusterId)
	if bizError != nil {
		logger.Warn("failed to submit audit log, query cluster error",
			zap.Uint64("clusterId", restoreReq.ClusterId), zap.Error(bizError))
		return
	}
	eventType := auditlog.EventRecoverFromBackup
	if cluster.GetDedicatedCluster().GetState() == dedicatedv1.Cluster_DELETED {
		eventType = auditlog.EventRecoverFromTrash
	}

	event := auditlog.NewEvent(ctx, eventType).
		SetResult(rErr == nil).
		SetOrgID(restoreReq.OrgId).
		SetProjectID(restoreReq.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
}

func (c *backupControllerImpl) buildCreateRestoreReq(
	ctx context.Context, req *dataflow.BackupRestoreReq,
) *domain.CreateRestoreReqV2 {
	clusterConfig := c.buildClusterConfig(req)
	user := account.GetUserFromContext(ctx)
	restoreReq := &domain.CreateRestoreReqV2{
		Mode:          domain.RestoreModeDataflow2Domain(req.RestoreMode),
		BackupID:      req.Id,
		ClusterConfig: *clusterConfig,
		UserID:        user.ID,
		ProjectID:     req.ProjectId,
		TenantID:      req.OrgId,
		ClusterID:     req.ClusterId,
		RestoredTs:    req.RestoredTs,
		BackupRegion:  req.BackupRegion,
	}
	return restoreReq
}

func (c *backupControllerImpl) buildClusterConfig(req *dataflow.BackupRestoreReq) *domain.ClusterConfig {
	clusterConfig := &domain.ClusterConfig{
		Region:       req.GetRegion(),
		Name:         req.GetClusterName(),
		RootPassword: req.GetPassword(),
		Port:         req.GetPort(),
		Components:   c.convertClusterComponents(req),
	}
	return clusterConfig
}

func (c *backupControllerImpl) convertClusterComponents(req *dataflow.BackupRestoreReq) *domain.ClusterComponents {
	clusterComponents := &domain.ClusterComponents{}
	pie.Of(req.Components).Each(func(component *dataflow.SimpleClusterComponent) {
		switch component.ComponentType {
		case dataflow.ComponentType_COMPONENT_TYPE_TIDB:
			clusterComponents.Tidb = &domain.TiDBComponent{
				NodeSize:     component.NodeProfileName,
				NodeQuantity: component.Replicas,
			}
		case dataflow.ComponentType_COMPONENT_TYPE_TIKV:
			clusterComponents.Tikv = &domain.TiKVComponent{
				NodeSize:       component.NodeProfileName,
				NodeQuantity:   component.Replicas,
				StorageType:    component.StorageType,
				DataStoreIops:  component.DataStoreIops,
				RaftIops:       component.RaftIops,
				StorageSizeGib: int64(component.StorageSizeGi),
			}
		case dataflow.ComponentType_COMPONENT_TYPE_TIFLASH:
			if component.Replicas == 0 {
				return
			}
			clusterComponents.Tiflash = &domain.TiFlashComponent{
				NodeSize:       component.NodeProfileName,
				NodeQuantity:   component.Replicas,
				StorageType:    component.StorageType,
				StorageSizeGib: int64(component.StorageSizeGi),
			}
		case dataflow.ComponentType_COMPONENT_TYPE_TIPROXY:
			clusterComponents.TiProxy = &domain.TiProxyComponent{
				NodeSize:     component.NodeProfileName,
				NodeQuantity: component.Replicas,
			}
		}
	})
	return clusterComponents
}

func (c *backupControllerImpl) checkRestoreReqIsValid(
	ctx context.Context, req *dataflow.BackupRestoreReq,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	restoreMode := domain.RestoreModeDataflow2Domain(req.RestoreMode)
	if err := utils.CheckParamEmptyError("restoreMode", restoreMode); err != nil {
		logger.Warn("invalid param", zap.Error(err))
		return err
	}
	if err := utils.CheckParamEmptyError("components", req.Components); err != nil {
		logger.Warn("invalid param", zap.Error(err))
		return err
	}
	if err := utils.CheckParamEmptyError("region", req.Region); err != nil {
		logger.Warn("invalid param", zap.Error(err))
		return err
	}
	if err := utils.CheckParamEmptyError("clusterName", req.ClusterName); err != nil {
		logger.Warn("invalid param", zap.Error(err))
		return err
	}
	if err := utils.CheckParamInvalidError("port", req.Port, func(port int32) bool {
		return port > 0 && port < 65535
	}); err != nil {
		logger.Warn("invalid param", zap.Error(err))
		return err
	}
	if err := c.checkRestoreSnapshotReqIsValid(ctx, req); err != nil {
		return err
	}
	if err := c.checkRestorePitReqIsValid(ctx, req); err != nil {
		return err
	}
	return nil
}

func (c *backupControllerImpl) checkRestoreSnapshotReqIsValid(
	ctx context.Context, req *dataflow.BackupRestoreReq,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	if req.RestoreMode != dataflow.RestoreMode_RESTORE_MODE_SNAPSHOT {
		return nil
	}
	if err := utils.CheckParamEmptyError("backupID", req.Id); err != nil {
		logger.Warn("invalid param", zap.Error(err))
		return err
	}
	return nil
}

func (c *backupControllerImpl) checkRestorePitReqIsValid(
	ctx context.Context, req *dataflow.BackupRestoreReq,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	if req.RestoreMode != dataflow.RestoreMode_RESTORE_MODE_PITR {
		return nil
	}
	if err := utils.CheckParamEmptyError("restoredTs", req.RestoredTs); err != nil {
		logger.Warn("invalid param", zap.Error(err))
		return err
	}
	return nil
}

func (c *backupControllerImpl) GetRestoreProgress(
	ctx context.Context, req *dataflow.RestoreProgressReq,
) (*dataflow.RestoreProgressResp, error) {
	ctx = log.WithTraceID(ctx)
	queryResp, err := c.restoreRouter.Query(ctx, &domain.QueryRestoreReqV2{
		Filter: &domain.ListRestoreFilter{
			TenantIDs:  []uint64{req.OrgId},
			ProjectIDs: []uint64{req.ProjectId},
			ClusterIDs: []uint64{req.ClusterId},
		},
	})
	if err != nil {
		return nil, err
	}

	status := DomainRestoreStatusToDataflow(queryResp.Status)
	resp := &dataflow.RestoreProgressResp{
		RestoreStatus: status,
	}
	switch status {
	case dataflow.RestoreStatus_RESTORE_STATUS_SUCCESS:
		resp.Progress = 100
		return resp, nil
	case dataflow.RestoreStatus_RESTORE_STATUS_RUNNING:
		resp.Progress = queryResp.Progress
		resp.Steps = c.convertRestoreStepToDataflow(queryResp)
		return resp, nil
	case dataflow.RestoreStatus_RESTORE_STATUS_FAILED:
		resp.RestoreError = &dataflow.RestoreErrorInfo{
			Retryable: false,
			Message:   "Restore failed, please contact administrator",
		}
		return resp, nil
	default:
		return resp, nil
	}
}

func (c *backupControllerImpl) convertRestoreStepToDataflow(
	resp *domain.QueryRestoreRespV2,
) []*dataflow.RestoreStepProgress {
	if resp == nil || len(resp.Steps) == 0 {
		return nil
	}
	return pie.Map(resp.Steps, func(p domain.RestoreStep) *dataflow.RestoreStepProgress {
		return &dataflow.RestoreStepProgress{
			StepName: p.StepName,
			Progress: p.Progress,
		}
	})
}

func (c *backupControllerImpl) CreateBackupSchedule(
	ctx context.Context, req *dataflow.BackupCreateReq,
) (_ *dataflow.CreateBackupScheduleResp, rErr error) {
	var backupID uint64
	defer func() {
		c.recordAuditLogForCreateBackup(ctx, req, backupID, rErr)
	}()

	_, rErr = c.checkClusterPermission(ctx, req.OrgId, req.ProjectId, req.ClusterId)
	if rErr != nil {
		return nil, rErr
	}

	user := account.GetUserFromContext(ctx)
	createReq := &domain.AsyncCreateBackupReq{
		Name:        req.Name,
		Description: req.Description,
		ProjectID:   req.ProjectId,
		ClusterID:   req.ClusterId,
		BackupType:  req.BackupType,
		UserID:      user.ID,
	}
	createResp, rErr := c.backupCreateRouter.AsyncCreate(ctx, createReq)
	if rErr != nil {
		return nil, rErr
	}
	if createResp == nil {
		return &dataflow.CreateBackupScheduleResp{}, nil
	}
	backupID = createResp.BackupID
	return &dataflow.CreateBackupScheduleResp{
		Id:       createResp.BackupScheduleID,
		BackupId: createResp.BackupID,
	}, nil
}

func (c *backupControllerImpl) recordAuditLogForCreateBackup(
	ctx context.Context, req *dataflow.BackupCreateReq, backupID uint64, rErr error,
) {
	if req.BackupType == dataflow.BackupType_MANUAL {
		logDetail := &CreateManualBackupAuditLogDetail{
			BackupID:   strconv.FormatUint(backupID, 10),
			BackupName: req.Name,
			Error:      c.ConvertAuditLogDetailErrMsg(rErr),
		}
		details, err := c.ConvertToAuditLogDetail(logDetail)
		if err != nil {
			log.GetLogger(ctx).Warn("failed to convert to audit log detail", zap.Error(err))
			return
		}
		event := auditlog.NewEvent(ctx, auditlog.EventDoManualBackup).
			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 {
			log.GetLogger(ctx).Warn("failed to submit audit log", zap.Error(err))
			return
		}
		return
	}
}

func (c *backupControllerImpl) recordAuditLogForDeleteBackup(
	ctx context.Context, req *dataflow.BackupReq, rErr error,
) {
	logger := log.GetLogger(ctx)
	resp, err := c.snapshotBackupModel.Query(ctx, &domain.QueryBackupReq{
		Filter: &domain.ListBackupFilter{
			BackupIDs: []uint64{req.Id},
		},
	})
	if err != nil || resp.BackupEntity == nil || resp.BackupEntity.BackupSchedule == nil {
		logger.Warn("failed to submit audit log, query backup error",
			zap.Uint64("backupID", req.Id), zap.Error(err))
		return
	}
	logDetail := &DeleteBackupAuditLogDetail{
		BackupName: resp.BackupEntity.BackupSchedule.Name,
		BackupType: string(resp.BackupEntity.BackupSchedule.Type),
		Error:      c.ConvertAuditLogDetailErrMsg(rErr),
	}
	details, ierr := c.ConvertToAuditLogDetail(logDetail)
	if ierr != nil {
		log.GetLogger(ctx).Warn("failed to convert to audit log detail", zap.Error(ierr))
		return
	}

	event := auditlog.NewEvent(ctx, auditlog.EventDeleteBackup).
		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
	}
}

func (c *backupControllerImpl) Delete(ctx context.Context, req *dataflow.BackupReq) (resp *emptypb.Empty, rErr error) {
	defer func() {
		c.recordAuditLogForDeleteBackup(ctx, req, rErr)
	}()
	logger := log.GetLogger(ctx)
	ctx = log.WithTraceID(ctx)

	if _, err := c.checkBackupPermission(ctx, req.OrgId, req.ProjectId, req.ClusterId, req.Id); err != nil {
		return &emptypb.Empty{}, err
	}

	if err := c.doBillingActionCheckForBackupDeletion(ctx, req.ClusterId, req.Id); err != nil {
		return &emptypb.Empty{}, err
	}

	deleteBackupReq := &domain.AsyncDeleteBackupReq{
		BackupID: req.Id,
	}

	_, rErr = c.snapshotBackupModel.AsyncDelete(ctx, deleteBackupReq)
	if rErr != nil {
		logger.Error("failed to delete backup of cluster",
			zap.Uint64("clusterID", req.ClusterId), zap.Uint64("backupID", req.Id), zap.Error(rErr))
		return &emptypb.Empty{}, rErr
	}

	return &emptypb.Empty{}, nil
}

func (c *backupControllerImpl) ListRecycleBinClusters(
	ctx context.Context, req *dataflow.ListRecycleBinClustersReq,
) (*dataflow.ListRecycleBinClustersResp, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	listReq := &domain.ListRecycleBinClustersReq{
		TenantID:  req.OrgId,
		ProjectID: req.ProjectId,
		Pager: &domain.Pager{
			Page:     int64(req.Page),
			PageSize: int64(req.PerPage),
		},
	}
	listResp, err := c.snapshotBackupModel.ListRecycleBinClusters(ctx, listReq)
	if err != nil {
		logger.Warn("list recycle bin clusters error", zap.Any("req", req), zap.Error(err))
		return nil, err
	}

	items := make([]*dataflow.RecycleBinCluster, 0, len(listResp.Items))
	for _, cluster := range listResp.Items {
		items = append(items, cluster.ToDataflowRecycleBinCluster())
	}
	return &dataflow.ListRecycleBinClustersResp{
		Items: items,
		Total: int32(listResp.Total),
	}, nil
}

func (c *backupControllerImpl) ListRecycleBinClustersV2(
	ctx context.Context,
	req *dataflow.ListRecycleBinClustersV2Req,
) (*dataflow.ListRecycleBinClustersV2Resp, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	listReq := &domain.ListRecycleBinClustersV2Req{
		TenantID:    req.OrgId,
		ProjectIDs:  req.ProjectIds,
		ClusterName: req.ClusterName,
		Regions:     req.Regions,
		Pager: &domain.Pager{
			Page:     int64(req.Page),
			PageSize: int64(req.PerPage),
		},
	}
	if req.StartTime != nil {
		startTime := req.StartTime.AsTime()
		listReq.StartTime = &startTime
	}
	if req.EndTime != nil {
		endTime := req.EndTime.AsTime()
		listReq.EndTime = &endTime
	}
	listResp, err := c.snapshotBackupModel.ListRecycleBinClustersV2(ctx, listReq)
	if err != nil {
		logger.Warn("list recycle bin clusters v2 error", zap.Any("req", req), zap.Error(err))
		return nil, err
	}

	items := make([]*dataflow.RecycleBinCluster, 0, len(listResp.Items))
	for _, cluster := range listResp.Items {
		items = append(items, cluster.ToDataflowRecycleBinCluster())
	}

	return &dataflow.ListRecycleBinClustersV2Resp{
		Items: items,
		Total: int32(listResp.Total),
	}, nil
}

func (c *backupControllerImpl) GetRecycleBinCluster(
	ctx context.Context, req *dataflow.GetRecycleBinClusterReq,
) (*dataflow.GetRecycleBinClusterResp, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	cluster, err := c.checkClusterPermission(ctx, req.OrgId, req.ProjectId, req.ClusterId)
	if err != nil {
		logger.Warn("get cluster error", zap.Any("req", req), zap.Error(err))
		return nil, err
	}
	dc := cluster.GetDedicatedCluster()
	if dc == nil || dc.GetDeleteTime() == nil {
		logger.Warn("cluster is not deleted", zap.Uint64("cluster_id", req.ClusterId))
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster is not deleted")
	}

	return &dataflow.GetRecycleBinClusterResp{
		Cluster: &dataflow.RecycleBinCluster{
			Id:         cluster.GetClusterId(),
			Name:       cluster.GetClusterName(),
			Provider:   cluster.GetCloudProvider(),
			Region:     cluster.GetRegion(),
			RegionName: dc.GetRegionDisplayName(),
			DeleteTime: dc.GetDeleteTime(),
		},
	}, nil
}

func (c *backupControllerImpl) DeleteDeletedClusterAndBackups(
	ctx context.Context, req *dataflow.ClusterIdReq,
) (*emptypb.Empty, error) {
	logger := log.GetLogger(ctx)
	ctx = log.WithTraceID(ctx)

	_, err := c.checkClusterPermission(ctx, req.OrgId, req.ProjectId, req.ClusterId)
	if err != nil {
		logger.Warn("get cluster error", zap.Any("req", req), zap.Error(err))
		return &emptypb.Empty{}, err
	}

	// Delete all backups for a deleted cluster don't have a particular backup id, so give the 0 to instead.
	if err := c.doBillingActionCheckForBackupDeletion(ctx, req.ClusterId, 0); err != nil {
		return &emptypb.Empty{}, err
	}

	backups, err := c.snapshotBackupModel.List(ctx,
		&domain.ListBackupReq{
			Filter: &domain.ListBackupFilter{
				ClusterIDs: []uint64{req.ClusterId},
			},
		})
	if err != nil {
		logger.Error("failed to get backups of deleted cluster",
			zap.Uint64("clusterID", req.ClusterId), zap.Error(err))
		return &emptypb.Empty{}, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterId)
	}
	for _, backup := range backups.Items {
		_, deleteErr := c.snapshotBackupModel.AsyncDelete(ctx,
			&domain.AsyncDeleteBackupReq{
				BackupID: backup.ID,
			})
		if deleteErr != nil {
			logger.Error("failed to deleted backup",
				zap.Uint64("clusterID", req.ClusterId), zap.Error(deleteErr), zap.Uint64("backupId", backup.ID))
			return &emptypb.Empty{}, deleteErr
		}
	}

	return &emptypb.Empty{}, nil
}

func (c *backupControllerImpl) List(ctx context.Context, req *dataflow.BackupListReq) (*dataflow.Backups, error) {
	ctx = log.WithTraceID(ctx)

	_, err := c.checkClusterPermission(ctx, req.OrgId, req.ProjectId, req.ClusterId)
	if err != nil {
		return nil, err
	}

	listBackupReq := c.buildListBackupReq(req)
	listBackupResp, err := c.snapshotBackupModel.List(ctx, listBackupReq)
	if err != nil {
		return nil, err
	}
	if listBackupResp == nil {
		return &dataflow.Backups{}, nil
	}
	if len(listBackupResp.Items) == 0 {
		return &dataflow.Backups{Total: uint32(listBackupResp.Total)}, nil
	}
	items := c.convertListBackupItemsResp(ctx, listBackupResp)
	return &dataflow.Backups{
		Total: uint32(listBackupResp.Total),
		Items: items,
	}, nil
}

func (c *backupControllerImpl) convertListBackupItemsResp(
	ctx context.Context, listBackupResp *domain.ListBackupResp,
) []*dataflow.Backup {
	items := pie.Map(listBackupResp.Items, func(item domain.ListBackupRespItem) *dataflow.Backup {
		return c.convertOneListBackupItemResp(ctx, item)
	})
	return items
}

func (c *backupControllerImpl) convertOneListBackupItemResp(
	ctx context.Context, item domain.ListBackupRespItem,
) *dataflow.Backup {
	logger := log.GetLogger(ctx)
	backup := &dataflow.Backup{
		Id: item.ID,
		Cluster: &dataflow.BackupCluster{
			ClusterId:         item.ClusterID,
			ClusterVersion:    item.ClusterVersion,
			ClusterComponents: c.convertClusterComponentsResp(&item.ClusterConfig),
		},
		Name:           item.Name,
		Description:    item.Description,
		CreatedAt:      timestamppb.New(item.CreatedTime),
		Size:           item.Size,
		Status:         domain.BackupStatusDomain2DataflowExternalStatus(item.Status),
		Type:           domain.BackupTypeDomain2Dataflow(item.Type),
		IsVolumeBackup: false,
	}

	if item.ExpirationTime != nil {
		backup.ExpirationTime = timestamppb.New(*item.ExpirationTime)
	}
	if item.BackupTs != nil {
		backup.BackupTso = util.TransToTSO(*item.BackupTs)
		backupTs := timestamppb.New(time.Unix(*item.BackupTs, 0))
		if err := backupTs.CheckValid(); err != nil {
			logger.Warn("failed to parse backup's BackupTs",
				zap.Uint64("clusterID", item.ClusterID), zap.Error(err))
		} else {
			backup.BackupTs = backupTs
		}
	}
	backup.Regions = pie.Map(item.Regions, func(region domain.RegionInfo) *dataflow.RegionInfo {
		return &dataflow.RegionInfo{
			Name:        region.Name,
			DisplayName: region.DisplayName,
		}
	})
	return backup
}

//nolint:gomnd
func (c *backupControllerImpl) convertClusterComponentsResp(
	clusterConfig *domain.ClusterConfig,
) *dataflow.ClusterComponents {
	if clusterConfig == nil || clusterConfig.Components == nil {
		return nil
	}
	clusterComponents := &dataflow.ClusterComponents{}
	if clusterConfig.Components.Tidb != nil {
		clusterComponents.Tidb = &dataflow.ClusterComponent{
			Replicas: uint32(clusterConfig.Components.Tidb.NodeQuantity),
			NodeProfile: &dataflow.NodeProfile{
				Compute: &dataflow.Compute{
					Cpu:      float32(clusterConfig.Components.Tidb.CpuNum),
					MemoryMi: uint32(clusterConfig.Components.Tidb.MemoryGib * 1024),
				},
				Name: clusterConfig.Components.Tidb.NodeSize,
			},
		}
	}
	if clusterConfig.Components.Tikv != nil {
		clusterComponents.Tikv = &dataflow.ClusterComponent{
			Replicas: uint32(clusterConfig.Components.Tikv.NodeQuantity),
			NodeProfile: &dataflow.NodeProfile{
				Compute: &dataflow.Compute{
					Cpu:      float32(clusterConfig.Components.Tikv.CpuNum),
					MemoryMi: uint32(clusterConfig.Components.Tikv.MemoryGib * 1024),
				},
				Name:      clusterConfig.Components.Tikv.NodeSize,
				StorageGi: uint32(clusterConfig.Components.Tikv.StorageSizeGib),
			},
		}
	}
	if clusterConfig.Components.Tiflash != nil {
		clusterComponents.Tiflash = &dataflow.ClusterComponent{
			Replicas: uint32(clusterConfig.Components.Tiflash.NodeQuantity),
			NodeProfile: &dataflow.NodeProfile{
				Compute: &dataflow.Compute{
					Cpu:      float32(clusterConfig.Components.Tiflash.CpuNum),
					MemoryMi: uint32(clusterConfig.Components.Tiflash.MemoryGib * 1024),
				},
				Name:      clusterConfig.Components.Tiflash.NodeSize,
				StorageGi: uint32(clusterConfig.Components.Tiflash.StorageSizeGib),
			},
		}
	}
	return clusterComponents
}

func (c *backupControllerImpl) buildListBackupReq(req *dataflow.BackupListReq) *domain.ListBackupReq {
	listBackupReq := &domain.ListBackupReq{
		Filter: &domain.ListBackupFilter{
			ClusterIDs: []uint64{req.ClusterId},
		},
	}
	if req.BackupId != nil {
		listBackupReq.Filter.BackupIDs = []uint64{*req.BackupId}
	}
	if req.Status != nil {
		listBackupReq.Filter.Statuses = []domain.BackupStatus{domain.BackupStatusDataflow2Domain(*req.Status)}
	}
	if req.Region != nil {
		listBackupReq.Filter.Region = req.Region
	}

	listBackupReq.Pager = domain.NewPager(domain.PageSizeOp(int64(req.PerPage)), domain.PageNumberOp(int64(req.Page)))
	return listBackupReq
}

func (c *backupControllerImpl) GetBackupOptions(
	ctx context.Context, req *dataflow.BackupOptionsReq,
) (*dataflow.BackupOptionsResp, error) {
	ctx = log.WithTraceID(ctx)

	_, err := c.checkClusterPermission(ctx, req.OrgId, req.ProjectId, req.ClusterId)
	if err != nil {
		return nil, err
	}

	queryResp, err := c.backupSettingModel.QueryBackupRegionOptions(ctx, &domain.QueryBackupRegionOptionsReq{
		ClusterID: req.ClusterId,
		TenantID:  req.OrgId,
		ProjectID: req.ProjectId,
	})
	if err != nil {
		return nil, err
	}

	// default info
	resp := &dataflow.BackupOptionsResp{
		ClusterId: queryResp.ClusterID,
		DualRegionOptions: &dataflow.BackupOptionsResp_DualRegionOptions{
			ClusterRegion: &dataflow.RegionInfo{
				Name:        queryResp.ClusterRegion.Name,
				DisplayName: queryResp.ClusterRegion.DisplayName,
			},
		},
	}
	if len(queryResp.BackupRegionOptions) == 0 {
		return resp, nil
	}

	pie.Of(queryResp.BackupRegionOptions).Each(func(item domain.BackupRegionOption) {
		option := &dataflow.BackupOptionsResp_RegionOptions{
			Continent:   item.Continent,
			Description: item.Description,
		}
		if item.Region != nil {
			option.Region = &dataflow.RegionInfo{
				Name:        item.Region.Name,
				DisplayName: item.Region.DisplayName,
			}
		}
		resp.DualRegionOptions.RegionOptions = append(resp.DualRegionOptions.RegionOptions, option)
	})
	return resp, nil
}

func (c *backupControllerImpl) ListPiTRTimeRanges(
	ctx context.Context, req *dataflow.PiTRTimeRangeListReq,
) (resp *dataflow.PiTRTimeRangeListResp, err error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).Named("backupControllerImpl.ListPiTRTimeRanges").
		With(zap.Any("cluster id", req.ClusterId))

	_, err = c.checkClusterPermission(ctx, req.OrgId, req.ProjectId, req.ClusterId)
	if err != nil {
		return nil, err
	}

	canEnable, err := c.canClusterVersionEnableLogBackup(ctx, req.OrgId, req.ProjectId, req.ClusterId)
	if err != nil {
		logger.Error("failed to check cluster version",
			zap.Uint64("clusterID", req.ClusterId), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterId)
	}
	if !canEnable {
		return &dataflow.PiTRTimeRangeListResp{
			Msg: fmt.Sprintf("cluster version should be at least %s to support PiTR",
				internal_config.Global().BackupConfig.LogBackupEnableMinVer),
		}, nil
	}

	listPitrReq := &domain.PiTRTimeRangeListReq{
		OrgId:     req.OrgId,
		ProjectId: req.ProjectId,
		ClusterId: req.ClusterId,
	}
	listResp, rerr := c.PitBackupModel.ListPiTRTimeRanges(ctx, listPitrReq)
	if rerr != nil {
		logger.Error("Fail to ListPiTRTimeRanges", zap.Error(rerr))
		return nil, rerr
	}
	if len(listResp.Items) == 0 {
		return &dataflow.PiTRTimeRangeListResp{Msg: listResp.Msg}, nil
	}
	items := c.convertPiTRTimeRangeListResp(ctx, listResp)
	resp = &dataflow.PiTRTimeRangeListResp{
		Items: items,
		Msg:   listResp.Msg,
	}
	return resp, rerr
}

func (c *backupControllerImpl) canClusterVersionEnableLogBackup(
	ctx context.Context, orgId uint64, projectId uint64, clusterId uint64,
) (bool, error) {
	logger := log.GetLogger(ctx).Named("snapshotBackupModel.canClusterVersionEnableLogBackup").
		With(zap.Any("cluster id", clusterId))
	cluster, bizErr := c.mgmtCli.GetDedicatedClusterBasicByID(ctx, clusterId)
	if bizErr != nil {
		logger.Sugar().Warnf("get cluster %d failed: %s", clusterId, bizErr.Error())
		return false, nil
	}

	clusterVersion := cluster.GetDedicatedCluster().GetVersion()
	v, err := semver.NewVersion(clusterVersion)
	if err != nil {
		logger.Sugar().Warnf("parse cluster %d version %s failed: %s", clusterId, clusterVersion, err.Error())
		return false, nil
	}

	// check tidb version and phase
	backupCfg := internal_config.Global().BackupConfig
	if !backupCfg.LogBackupEnableMinVerConstraint.Check(v) {
		logger.Sugar().Warnf("log backup enable min version is %s, but cluster %d version is %s",
			backupCfg.LogBackupEnableMinVer, clusterId, clusterVersion)
		return false, nil
	}
	return true, nil
}

func (c *backupControllerImpl) convertPiTRTimeRangeListResp(
	ctx context.Context, resp *domain.PiTRTimeRangeListResp,
) []*dataflow.PiTRRegionTimeRange {
	items := pie.Map(resp.Items, func(item *domain.PiTRRegionTimeRange) *dataflow.PiTRRegionTimeRange {
		return c.convertOnePiTRTimeRangeResp(ctx, item)
	})
	return items
}

func (c *backupControllerImpl) convertOnePiTRTimeRangeResp(
	ctx context.Context, item *domain.PiTRRegionTimeRange,
) *dataflow.PiTRRegionTimeRange {
	timeRanges := pie.Map(item.TimeRanges, func(timeRange *domain.PiTRTimestampRange) *dataflow.PiTRTimeRange {
		return &dataflow.PiTRTimeRange{
			StartTs: timeRange.StartTS,
			EndTs:   timeRange.EndTS,
		}
	})
	piTRTimeRange := &dataflow.PiTRRegionTimeRange{
		Region: &dataflow.RegionInfo{
			Name:        item.Region.Name,
			DisplayName: item.Region.DisplayName,
		},
		TimeRanges: timeRanges,
	}
	return piTRTimeRange
}

func (c *backupControllerImpl) doBillingActionCheckForBackupDeletion(
	ctx context.Context, clusterId, backupID uint64,
) error {
	logger := log.GetLogger(ctx)

	cluster, err := c.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, clusterId)
	if err != nil {
		logger.Warn("get cluster error", zap.Uint64("clusterId", clusterId), zap.Error(err))
		return err
	}

	checkResult, berr := c.billingHandler.ActionCheck(ctx, cluster.GetOrganizationId(), billing.ActionInfo{
		Resource: billing.Resource{
			Type:       billing.ResourceTypeDedicatedBackup,
			InstanceID: fmt.Sprint(backupID),
			Region:     cluster.GetRegion(),
			Provider:   cluster.GetCloudProvider(),
			IsFree:     false,
		},
		Action: billing.ActionDelete,
	})
	if berr != nil {
		logger.Info("failed to check billing action", zap.Error(berr))
		return berr
	}
	if !checkResult.Pass {
		return bizerr_wrapper.NewInvalidInputError("billing action check failed: %s", pointer.Get(checkResult.Reason))
	}

	logger.Info("billing action check passed")
	return nil
}

func (c *backupControllerImpl) checkBackupPermission(
	ctx context.Context, tenantID, projectID, clusterID, backupID uint64,
) (*entity.Backup, error) {
	backup, err := c.backupRepo.Load(ctx, backupID)
	if err != nil {
		return nil, errors.Annotate(err, "get backup")
	}
	if backup.ClusterID.Uint64() != clusterID {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "backup id doesn't match cluster's information")
	}
	if backup.TenantID.Uint64() != tenantID || backup.ProjectID.Uint64() != projectID {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "org id or project id doesn't match cluster's information")
	}
	return backup, nil
}

func (c *backupControllerImpl) checkClusterPermission(
	ctx context.Context, tenantID, projectID, clusterID uint64,
) (*mgmt.Cluster, error) {
	logger := log.GetLogger(ctx)
	cluster, err := c.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, clusterID)
	if err != nil {
		return nil, fmt.Errorf("get cluster %d: %w", clusterID, err)
	}
	if cluster.GetProjectId() != projectID || cluster.GetOrganizationId() != tenantID {
		logger.Warn("cluster and request are not matched",
			zap.Any("cluster_id", clusterID),
			zap.Any("project_id", projectID),
			zap.Any("tenant_id", tenantID),
		)
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster is not found")
	}
	return cluster, nil
}