package server

import (
	"context"
	"fmt"
	"strings"

	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/server/br"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	braction "github.com/tidbcloud/cluster-service-ng/domain/workflow/action/br"
	"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/tioperator"
	"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"
)

func (s *serverImpl) CreateLightweightRestore(ctx context.Context, req *br.CreateLightweightRestoreInput) error {
	restoreAdaptor := br.GetLightweightRestoreAdaptor(req, s.clustergatewayCli, s.sharedPoolRepo)
	dagMeta := restoreAdaptor.GetCreationDagMeta(req.RestoreId, req.TargetSharedPoolId)
	targetSharedPool, err := s.sharedPoolRepo.Load(ctx, req.TargetSharedPoolId, false)
	if err != nil {
		return fmt.Errorf("failed to load targetSharedPool: %w", err)
	}
	if err := guardSharedPoolMutation(targetSharedPool, sharedPoolMutationBlockedWhilePaused); err != nil {
		return err
	}

	// prepare parameters
	targetKeyspace, err := s.resolveClusterKeyspace(ctx, req.TargetClusterId, req.TargetSharedPoolId)
	if err != nil {
		return fmt.Errorf("get targetKeyspace for cluster %s failed: %w", req.TargetClusterId, err)
	}
	sourceKeyspace, err := s.resolveClusterKeyspace(ctx, req.SourceClusterId, req.SourceSharedPoolId)
	if err != nil {
		return fmt.Errorf("get sourceKeyspace for cluster %s failed: %w", req.SourceClusterId, err)
	}
	targetPdCluster, err := s.clustergatewayCli.GetCluster(ctx, targetSharedPool)
	if err != nil {
		return fmt.Errorf("failed to get target pd cluster id: %w", err)
	}

	// query existed restore record
	restore, err := s.lwRestoreRepo.Load(ctx, req.RestoreId, true)
	if err != nil && status.Code(err) != codes.NotFound {
		return fmt.Errorf("failed to load lw restore: %w", err)
	}
	if restore != nil {
		if restore.DeletedAt.Valid {
			return fmt.Errorf("restore id conflict with a deleted restore, please use a new restore id")
		}
		return fmt.Errorf("restore %s already exists, please don't create it again", req.RestoreId)
	}

	// create new dag instance if it's not exist
	cpu, memory, storage, err := s.getRestoreResource(ctx, req)
	if err != nil {
		return fmt.Errorf("failed to get restore resource: %w", err)
	}

	tilog.Info(ctx, "create a dag instance for lightweight restore...", zap.Any("dagMeta", dagMeta))
	workflowArgs := actions.ActionArgs{
		utils.TraceIdKey: utils.GetTraceId(ctx),
		// common
		workflow.RestoreIdKey:          actions.ActionArgString(req.RestoreId),
		workflow.SharedPoolIdKey:       actions.ActionArgString(req.TargetSharedPoolId),
		workflow.SourceSharedPoolIdKey: actions.ActionArgString(req.SourceSharedPoolId),
		// create tibr action
		workflow.TiBRMemoryKey:  actions.ActionArgString(memory),
		workflow.TiBRCPUKey:     actions.ActionArgString(cpu),
		workflow.TiBRStorageKey: actions.ActionArgString(storage),
	}

	restoreRecord := schema.LightweightRestore{
		RestoreID: req.RestoreId,
		Status:    schema.LightweightRestoreStatusRunning,

		SourceSharedPoolID: req.SourceSharedPoolId,
		TargetSharedPoolID: req.TargetSharedPoolId,
		SourceClusterID:    req.SourceClusterId,
		TargetClusterID:    req.TargetClusterId,
		TargetPdClusterID:  fmt.Sprintf("%d", targetPdCluster.Id),
		TargetKeyspace:     targetKeyspace,
		SourceKeyspace:     sourceKeyspace,
		KernelBackupID: func() string {
			if req.KernelBackupId != 0 {
				return fmt.Sprintf("%d", req.KernelBackupId)
			}
			return ""
		}(),
		KernelBackupName:    req.KernelBackupName,
		BackupDataSizeBytes: req.BackupDataSizeBytes,
		PointInTime:         req.PointInTime,
	}

	var dagExecId string
	var dagSubmitted bool
	err = s.transactioner.Transaction(ctx, func(ctx context.Context) error {
		dagExecId = dagMeta.DagInstanceName
		ierr := s.dagExecutor.Submit(ctx, dagMeta.DagName, workflowArgs, dagExecId)
		if ierr != nil {
			tilog.Info(ctx, "failed to submit dag instance for lightweight restore", zap.Any("dagMeta", dagMeta))
			return fmt.Errorf("failed to submit dag, dagName: %s, dagInstanceName: %s, err: %w", dagMeta.DagName, dagExecId, ierr)
		}
		dagSubmitted = true
		// save restore record
		if ierr = s.lwRestoreRepo.Create(ctx, &entity.LightweightRestore{LightweightRestore: restoreRecord}); ierr != nil {
			return fmt.Errorf("failed to save restore record: %w", ierr)
		}
		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 lightweight restore: %w", err)
	}

	return nil
}

type restoreResourceInput interface {
	GetResourceCpu() string
	GetResourceMemory() string
	GetResourceStorage() string
	GetTargetSharedPoolId() string
}

func (s *serverImpl) getRestoreResource(ctx context.Context, req restoreResourceInput) (cpu, memory, storage string, err error) {
	cpu, memory, storage = strings.TrimSpace(req.GetResourceCpu()), strings.TrimSpace(req.GetResourceMemory()), strings.TrimSpace(req.GetResourceStorage())

	if cpu != "" && memory != "" && storage != "" {
		return cpu, memory, storage, nil
	}

	if cpu == "" || memory == "" {
		// get cpu memory from tikv groups
		sharedPool, err := s.sharedPoolRepo.Load(ctx, req.GetTargetSharedPoolId(), false)
		if err != nil {
			return "", "", "", fmt.Errorf("failed to load sharedPool: %w", err)
		}
		remoteCli, err := s.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
		if err != nil {
			return "", "", "", fmt.Errorf("failed to get remote client for shared pool: %w", err)
		}
		tikgs, err := tioperator.ListTiKVGroups(ctx, remoteCli, req.GetTargetSharedPoolId())
		if err != nil {
			return "", "", "", fmt.Errorf("failed to list tikv groups: %w", err)
		}
		if len(tikgs) == 0 {
			return "", "", "", fmt.Errorf("no tikv group found for shared pool: %s", req.GetTargetSharedPoolId())
		}
		tikvGroup := tikgs[0]
		if tikvGroup.GetNodeQuota() == nil || tikvGroup.GetNodeQuota().GetNodeSpec() == nil {
			return "", "", "", fmt.Errorf("tikv group %s has no node spec", tikvGroup.GetId())
		}
		if cpu == "" {
			cpu = fmt.Sprintf("%d", tikvGroup.GetNodeQuota().GetNodeSpec().GetCpu())
		}
		if memory == "" {
			memory = fmt.Sprintf("%dGi", tikvGroup.GetNodeQuota().GetNodeSpec().GetMemoryGi())
		}
	}

	// get storage from default config
	storage = braction.DefaultTiBRStorageSize
	if s.bizCfg != nil && s.bizCfg.DefaultTiBRConfig.StorageSize != "" {
		storage = s.bizCfg.DefaultTiBRConfig.StorageSize
	}

	return cpu, memory, storage, nil
}