package server

import (
	"context"
	"fmt"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/schema"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type CreateManualBackupRestoreInput struct {
	*apiv1.CreateManualBackupRestoreRequest
}

func (s *serverImpl) CreateManualBackupRestore(ctx context.Context, input *CreateManualBackupRestoreInput) error {
	backup, err := s.manualBackupRepo.Load(ctx, input.SourceClusterId, input.BackupId, false)
	if err != nil {
		return fmt.Errorf("failed to load source manual backup: %w", err)
	}
	if backup.Status != schema.ManualBackupStatusSucceeded {
		return fmt.Errorf("source manual backup is not in succeeded status, current status: %s", backup.Status)
	}

	existingRestore, err := s.manualBackupRestoreRepo.Load(ctx, input.RestoreId, true)
	if err != nil && status.Code(err) != codes.NotFound {
		return fmt.Errorf("failed to load manual backup restore: %w", err)
	}
	if existingRestore != nil {
		if existingRestore.DeletedAt.Valid {
			return fmt.Errorf("restore id conflict with a deleted restore, please use a new restore id")
		}
		return nil
	}
	if err := s.guardSharedPoolMutationByID(ctx, input.TargetSharedPoolId, sharedPoolMutationBlockedWhilePaused); err != nil {
		return err
	}

	targetKeyspace, err := s.resolveClusterKeyspace(ctx, input.TargetClusterId, input.TargetSharedPoolId)
	if err != nil {
		return fmt.Errorf("get targetKeyspace for cluster %s failed: %w", input.TargetClusterId, err)
	}

	targetSharedPool, err := s.sharedPoolRepo.Load(ctx, input.TargetSharedPoolId, false)
	if err != nil {
		return fmt.Errorf("failed to load target sharedPool %s: %w", input.TargetSharedPoolId, err)
	}

	targetPdCluster, err := s.clustergatewayCli.GetCluster(ctx, targetSharedPool)
	if err != nil {
		return fmt.Errorf("failed to get target pd cluster id: %w", err)
	}

	cpu, memory, storage, err := s.getRestoreResource(ctx, input)
	if err != nil {
		return fmt.Errorf("failed to get restore resource: %w", err)
	}

	sourceSharedPoolID := backup.SharedPoolID
	isCrossSharedPool := sourceSharedPoolID != input.TargetSharedPoolId

	restoreRecord := schema.ManualBackupRestore{
		RestoreID:          input.RestoreId,
		Status:             schema.ManualBackupRestoreStatusRunning,
		SourceClusterID:    input.SourceClusterId,
		SourceBackupID:     input.BackupId,
		SourceSharedPoolID: sourceSharedPoolID,
		TargetSharedPoolID: input.TargetSharedPoolId,
		TargetClusterID:    input.TargetClusterId,
		TargetPdClusterID:  fmt.Sprintf("%d", targetPdCluster.Id),
		TargetKeyspace:     targetKeyspace,
		BackupMetaPath:     backup.CopyMetaPath,
		RestoreSize:        backup.RestoreSize,
	}

	workflowArgs := actions.ActionArgs{
		utils.TraceIdKey:               utils.GetTraceId(ctx),
		workflow.ManualRestoreIdKey:    actions.ActionArgString(input.RestoreId),
		workflow.SharedPoolIdKey:       actions.ActionArgString(input.TargetSharedPoolId),
		workflow.SourceSharedPoolIdKey: actions.ActionArgString(sourceSharedPoolID),
		workflow.TiBRCPUKey:            actions.ActionArgString(cpu),
		workflow.TiBRMemoryKey:         actions.ActionArgString(memory),
		workflow.TiBRStorageKey:        actions.ActionArgString(storage),
	}

	dagPlan := dag.PlanCreateManualBackupRestore
	if isCrossSharedPool {
		dagPlan = dag.PlanCreateManualBackupXRestore
	}

	dagExecId := fmt.Sprintf("manual-backup-restore-%s", input.RestoreId)
	if isCrossSharedPool {
		dagExecId = fmt.Sprintf("manual-backup-x-restore-%s", input.RestoreId)
	}
	var dagSubmitted bool
	err = s.transactioner.Transaction(ctx, func(ctx context.Context) error {
		if ierr := s.manualBackupRestoreRepo.Create(ctx, &entity.ManualBackupRestore{ManualBackupRestore: restoreRecord}); ierr != nil {
			return fmt.Errorf("failed to save manual backup restore record: %w", ierr)
		}

		if ierr := s.dagExecutor.Submit(ctx, dagPlan, workflowArgs, dagExecId); ierr != nil {
			tilog.Info(ctx, "failed to submit dag instance for manual backup restore", zap.Any("input", input))
			return fmt.Errorf("failed to submit dag, dagName: %s, dagInstanceName: %s, err: %w", dagPlan, dagExecId, ierr)
		}
		dagSubmitted = true
		return nil
	})
	if err != nil {
		if dagSubmitted {
			if abortErr := s.dagExecutor.Abort(ctx, dagExecId); abortErr != nil {
				tilog.Error(ctx, "failed to abort dag after transaction failure",
					zap.String("dagExecId", dagExecId),
					zap.Error(abortErr))
			}
		}
		return fmt.Errorf("failed to create manual backup restore: %w", err)
	}

	return nil
}