package server

import (
	"context"
	"encoding/base64"
	"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 CreateClassicRestoreInput struct {
	*apiv1.CreateClassicRestoreRequest
}

func (s *serverImpl) CreateClassicRestore(ctx context.Context, input *CreateClassicRestoreInput) error {
	if input.SourceRoleArn != "" && (input.AccessKeyId != "" || input.AccessKeySecret != "") {
		return status.Error(codes.InvalidArgument, "source role and access key credentials are mutually exclusive")
	}
	if (input.SourceRoleArn == "") != (input.SourceRoleExternalId == "") {
		return status.Error(codes.InvalidArgument, "source role ARN and external ID must be provided together")
	}
	// query existed restore record
	restore, err := s.classicRestoreRepo.Load(ctx, input.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", input.RestoreId)
	}
	if err := s.guardSharedPoolMutationByID(ctx, input.TargetSharedPoolId, sharedPoolMutationBlockedWhilePaused); err != nil {
		return err
	}

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

	tilog.Info(ctx, "create a dag instance for classic restore...", zap.Any("input", input))
	workflowArgs := actions.ActionArgs{
		utils.TraceIdKey:         utils.GetTraceId(ctx),
		workflow.RestoreIdKey:    actions.ActionArgString(input.RestoreId),
		workflow.SharedPoolIdKey: actions.ActionArgString(input.TargetSharedPoolId),
		workflow.TiBRMemoryKey:   actions.ActionArgString(memory),
		workflow.TiBRCPUKey:      actions.ActionArgString(cpu),
	}

	restoreRecord := schema.ClassicRestore{
		RestoreID: input.RestoreId,
		Status:    schema.ClassicRestoreStatusRunning,

		SourceBackupStorage:  input.BackupStorage,
		SourceRegion:         input.BackupRegion,
		BackupDataSizeBytes:  input.BackupDataSizeBytes,
		BackupCloudProvider:  input.BackupCloudProvider,
		SourceRoleARN:        input.SourceRoleArn,
		SourceRoleExternalID: input.SourceRoleExternalId,
		TargetSharedPoolID:   input.TargetSharedPoolId,
		TargetClusterID:      input.TargetClusterId,
	}
	if input.AccessKeyId != "" && input.AccessKeySecret != "" {
		restoreRecord.ClassicBackupAuth = &schema.ClassicBackupAuth{
			AccessKeyId:     base64.StdEncoding.EncodeToString([]byte(input.AccessKeyId)),
			AccessKeySecret: base64.StdEncoding.EncodeToString([]byte(input.AccessKeySecret)),
		}
	}

	var dagExecId string
	var dagSubmitted bool
	err = s.transactioner.Transaction(ctx, func(ctx context.Context) error {
		if ierr := s.classicRestoreRepo.Create(ctx, &entity.ClassicRestore{ClassicRestore: restoreRecord}); ierr != nil {
			return fmt.Errorf("failed to save classic restore record: %w", ierr)
		}

		dagExecId = fmt.Sprintf("cl-restore-%s", input.RestoreId)
		if ierr := s.dagExecutor.Submit(ctx, dag.PlanCreateClassicRestore, workflowArgs, dagExecId); ierr != nil {
			tilog.Info(ctx, "failed to submit dag instance for classic restore", zap.Any("input", input))
			return fmt.Errorf("failed to submit dag, dagName: %s, dagInstanceName: %s, err: %w", dag.PlanCreateClassicRestore, 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 classic restore: %w", err)
	}
	return nil
}