package backup

import (
	"context"

	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/dataflow-service/app/container"
	"github.com/tidbcloud/dataflow-service/app/controllers/console"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/model/backup"
	"github.com/tidbcloud/dataflow-service/internal/service"
	"github.com/tidbcloud/dataflow-service/internal/service/backup/cvt"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	pkggrpc "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	pkglog "github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tiinf/sdk/auditlog"
	"go.opentelemetry.io/otel"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
)

var Tracer = otel.Tracer("central/service/backup")

// Same as console
const (
	RestoreModeSnapshot = "Snapshot"
	RestoreModePITR     = "Point-in-Time"
)

type backupService struct {
	dataflow.UnimplementedBackupServiceServer
	service.OperationsService
	backupModel backup.Model
	mgmtCli     pkggrpc.MgmtSvcGrpcClient
}

// NewService constructs a backupService
func NewService(s service.OperationsService, model backup.Model, mgmtCli pkggrpc.MgmtSvcGrpcClient) dataflow.BackupServiceServer {
	return &backupService{OperationsService: s, backupModel: model, mgmtCli: mgmtCli}
}

func (s *backupService) userBackup(ctx context.Context) backup.UserContext {
	user := s.CurrentUser(ctx)
	return s.backupModel.User(ctx, user)
}

func (s *backupService) List(ctx context.Context, req *dataflow.BackupListReq) (_ *dataflow.Backups, err error) {
	ctx = pkglog.WithTraceID(ctx)
	return container.Global().BackupController.List(ctx, req)
}

func (s *backupService) ListLogBackups(ctx context.Context, req *dataflow.LogBackupListReq) (resp *dataflow.LogBackupListResp, err error) {
	ctx, span := Tracer.Start(ctx, "backupService.ListLogBackups")
	defer func() { tracing.EndSpan(span, err) }()
	userCtx := s.userBackup(ctx)
	if err = s.checkUserClusterPermission(ctx, userCtx, req.OrgId, req.ProjectId, req.ClusterId); err != nil {
		return nil, errors.Annotate(err, "check cluster permission")
	}

	dfLogBackups, err := s.listLogBackupsByClusterID(ctx, userCtx, req.ClusterId)
	if err != nil {
		return nil, errors.Annotate(err, "list log backups")
	}

	resp = &dataflow.LogBackupListResp{
		Items: service.Pagination(req.Page, req.PerPage, dfLogBackups).([]*dataflow.LogBackup),
		Total: uint32(len(dfLogBackups)),
	}
	return
}

// GetBackupOptions backup bucket setting
func (s *backupService) GetBackupOptions(ctx context.Context, req *dataflow.BackupOptionsReq) (_ *dataflow.BackupOptionsResp, err error) {
	return container.Global().BackupController.GetBackupOptions(ctx, req)
}

// CreateBackupSchedule creates a schedule for manual backup, if you want to set auto backup, please use ModifyBackupSetting
func (s *backupService) CreateBackupSchedule(ctx context.Context, req *dataflow.BackupCreateReq) (_ *dataflow.CreateBackupScheduleResp, rErr error) {
	var beforeSetting *dataflow.AutoBackupSetting
	defer func() {
		s.recordAuditLogForCreateBackup(ctx, req, beforeSetting, rErr)
	}()

	return container.Global().BackupController.CreateBackupSchedule(ctx, req)
}

func (s *backupService) recordAuditLogForCreateBackup(ctx context.Context, req *dataflow.BackupCreateReq, before *dataflow.AutoBackupSetting, rErr error) {
	if req.BackupType == dataflow.BackupType_AUTO {
		logger := log.WithContextS(ctx)
		auditLogDetails := console.ServerlessBackupSettingAuditLogDetail{
			After: &console.ServerlessBackupSetting{
				BackupTime:      req.BackupTime,
				BackupRetention: int(req.RetentionPeriod),
				BackupCycle:     "EveryDay",
			},
		}
		if rErr != nil {
			auditLogDetails.Error = rErr.Error()
		}
		if before != nil {
			auditLogDetails.Before = &console.ServerlessBackupSetting{
				BackupTime:      before.BackupTime,
				BackupRetention: int(before.RetentionPeriod),
				BackupCycle:     "EveryDay",
			}
		}
		details, err := console.ConvertToAuditLogDetail(auditLogDetails)
		if err != nil {
			logger.Warn("failed to build audit log detail", zap.Error(err))
			return
		}
		err = auditlog.NewEvent(ctx, auditlog.EventSetAutoBackup).
			SetResult(rErr == nil).
			SetOrgID(req.OrgId).
			SetProjectID(req.ProjectId).
			SetClusterID(req.ClusterId).
			SetDetails(details.Build()).
			Submit()
		if err != nil {
			logger.Warn("failed to submit audit log", zap.Error(err))
			return
		}
		return
	}
}

// Delete marks a backup as deleted.
// A record of the backup will still exist.
func (s *backupService) Delete(ctx context.Context, req *dataflow.BackupReq) (resp *emptypb.Empty, rErr error) {
	userContext := s.userBackup(ctx)

	cluster, err := s.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, req.ClusterId)
	if err != nil {
		rErr = err
		return
	}
	rErr = s.checkUserClusterPermissionV2(ctx, userContext, req.OrgId, req.ProjectId, cluster)
	if rErr != nil {
		return
	}

	return container.Global().BackupController.Delete(ctx, req)
}

func (s *backupService) ListPiTRTimeRanges(ctx context.Context, req *dataflow.PiTRTimeRangeListReq) (resp *dataflow.PiTRTimeRangeListResp, err error) {
	ctx, span := Tracer.Start(ctx, "backupService.ListPiTRTimeRanges")
	defer func() { tracing.EndSpan(span, err) }()
	logger := log.WithContextL(ctx)
	logger.Sugar().Debugf("start")

	return container.Global().BackupController.ListPiTRTimeRanges(ctx, req)
}

func (s *backupService) GetRestoreProgress(ctx context.Context, req *dataflow.RestoreProgressReq) (resp *dataflow.RestoreProgressResp, err error) {
	ctx, span := Tracer.Start(ctx, "backupService.GetRestoreProgress")
	defer func() { tracing.EndSpan(span, err) }()
	logger := log.WithContextL(ctx)
	logger.Sugar().Debugf("start")
	userCtx := s.userBackup(ctx)
	cluster, err := s.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, req.ClusterId)
	if err != nil {
		return nil, errors.Annotate(err, "get cluster")
	}

	if err = s.checkUserClusterPermissionV2(ctx, userCtx, req.OrgId, req.ProjectId, cluster); err != nil {
		return nil, errors.Annotate(err, "check cluster permission")
	}

	restore, err := userCtx.GetRestoreByCluster(ctx, entity.ID(req.ClusterId))
	if err != nil {
		return nil, errors.Annotate(err, "get restore")
	}

	if restore.KernelMetrics != nil {
		logger.Info("restore with workflow", zap.Any("cluster id", req.ClusterId))
		return container.Global().BackupController.GetRestoreProgress(ctx, req)
	}

	return &dataflow.RestoreProgressResp{}, nil
}

// Restore restores a backup to a new cluster
func (s *backupService) Restore(ctx context.Context, restoreReq *dataflow.BackupRestoreReq) (restoreResp *dataflow.RestoreResp, rErr error) {
	return container.Global().BackupController.Restore(ctx, restoreReq)
}

func (s *backupService) DeleteDeletedClusterAndBackups(
	ctx context.Context, req *dataflow.ClusterIdReq,
) (*emptypb.Empty, error) {
	return container.Global().BackupController.DeleteDeletedClusterAndBackups(ctx, req)
}

func (s *backupService) checkUserClusterPermission(ctx context.Context, userCtx backup.UserContext, tenantID, projectID, clusterID uint64) error {
	if err := userCtx.CheckOrgIDAndProjectID(entity.ID(tenantID), entity.ID(projectID)); err != nil {
		return err
	}
	cluster, err := s.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, clusterID)
	if err != nil {
		return errors.Annotate(err, "get cluster")
	}
	if cluster.OrganizationId != tenantID || cluster.ProjectId != projectID {
		return errcode.NewForbiddenErr(errors.New("org id or project id doesn't match cluster's information"))
	}
	return nil
}

func (s *backupService) checkUserClusterPermissionV2(
	ctx context.Context, userCtx backup.UserContext, tenantID, projectID uint64, cluster *mgmt.Cluster,
) error {
	if err := userCtx.CheckOrgIDAndProjectID(entity.ID(tenantID), entity.ID(projectID)); err != nil {
		return err
	}
	if cluster.OrganizationId != tenantID || cluster.ProjectId != projectID {
		return errcode.NewForbiddenErr(errors.New("org id or project id doesn't match cluster's information"))
	}
	return nil
}

func (s *backupService) listLogBackupsByClusterID(ctx context.Context, userCtx backup.UserContext, clusterID uint64) ([]*dataflow.LogBackup, error) {
	logBackups, err := userCtx.GetLogBackupsByCluster(ctx, entity.ID(clusterID))
	if err != nil {
		if errors.IsNotFound(err) {
			return []*dataflow.LogBackup{}, nil
		}
		return nil, errors.Annotate(err, "list log backups")
	}

	dfLogBackups := make([]*dataflow.LogBackup, 0, len(logBackups))
	for _, logBackup := range logBackups {
		dfLogBackup, err := cvt.LogBackupEntity2Dataflow(&logBackup)
		if err != nil {
			return nil, errors.Annotatef(err, "convert log backup %d", logBackup.ID)
		}
		dfLogBackups = append(dfLogBackups, dfLogBackup)
	}

	return dfLogBackups, nil
}

// Deprecated: remove it later
func (s *backupService) ListServerlessRestores(ctx context.Context, req *dataflow.ListServerlessRestoresReq) (resp *dataflow.ListServerlessRestoresResp, rErr error) {
	return nil, bizerr_wrapper.NewInternalErr("serverless backup is no longer supported by dataflow service")
}

// Deprecated: remove it later
func (s *backupService) ListServerlessPiTRTimeRanges(ctx context.Context, req *dataflow.ListServerlessPiTRTimeRangesReq) (resp *dataflow.ListServerlessPiTRTimeRangesResp, rErr error) {
	return nil, bizerr_wrapper.NewInternalErr("serverless backup is no longer supported by dataflow service")
}

func (c *backupService) ListRecycleBinClusters(
	ctx context.Context, req *dataflow.ListRecycleBinClustersReq,
) (*dataflow.ListRecycleBinClustersResp, error) {
	return container.Global().BackupController.ListRecycleBinClusters(ctx, req)
}

func (c *backupService) ListRecycleBinClustersV2(
	ctx context.Context, req *dataflow.ListRecycleBinClustersV2Req,
) (*dataflow.ListRecycleBinClustersV2Resp, error) {
	return container.Global().BackupController.ListRecycleBinClustersV2(ctx, req)
}

func (c *backupService) GetRecycleBinCluster(
	ctx context.Context, req *dataflow.GetRecycleBinClusterReq,
) (*dataflow.GetRecycleBinClusterResp, error) {
	return container.Global().BackupController.GetRecycleBinCluster(ctx, req)
}