package server

import (
	"context"
	"errors"
	"fmt"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"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/runtime/executorext"
	"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"
)

type DeleteClassicRestoreInput struct {
	*apiv1.DeleteClassicRestoreRequest
}

func (s *serverImpl) DeleteClassicRestore(ctx context.Context, input *DeleteClassicRestoreInput) error {
	// check if restore is already done
	restore, err := s.classicRestoreRepo.Load(ctx, input.RestoreId, true)
	if err != nil {
		return fmt.Errorf("failed to load classic restore: %w", err)
	}
	// DeletedAt is only set after the restore dag is done
	if restore.DeletedAt.Valid {
		tilog.Info(ctx, "classic restore is already deleted, no need to create deletion dag", zap.String("restoreId", input.RestoreId))
		return nil
	}
	if err := s.guardSharedPoolMutationByID(ctx, restore.TargetSharedPoolID, sharedPoolMutationBlockedWhilePaused); err != nil {
		return err
	}

	filters := []executorext.DagInsFilter{
		{
			ArgKey:    workflow.RestoreIdKey,
			ArgValues: []string{input.RestoreId},
		},
	}

	// Cancel create dag if exist
	creationDagIns, err := s.dagInstanceFinder.GetRunningDagInstance(ctx, dag.PlanCreateClassicRestore, filters...)
	if err != nil && !errors.Is(err, executorext.DagInsNotFoundErr) {
		return err
	}
	if creationDagIns != nil {
		err = s.transactioner.Transaction(ctx, func(ctx context.Context) error {
			ierr := s.dagExecutor.Abort(ctx, creationDagIns.ExecTask.ExecId)
			if ierr != nil {
				tilog.Error(ctx, fmt.Sprintf("abort existed %s dag failed", dag.PlanCreateClassicRestore),
					zap.Error(ierr),
					zap.String("restoreId", input.RestoreId),
					zap.Any("dag ins", utils.JSON(creationDagIns)))
				return fmt.Errorf("abort existed %s dag failed: %w", dag.PlanCreateClassicRestore, ierr)
			}
			ierr = s.classicRestoreRepo.SetStatus(ctx, input.RestoreId, schema.ClassicRestoreStatusCancelled)
			if ierr != nil {
				tilog.Error(ctx, "mark db classic restore record as cancelled failed",
					zap.Error(ierr),
					zap.String("restoreId", input.RestoreId))
				return fmt.Errorf("mark db classic restore record as cancelled: %w", ierr)
			}
			return nil
		})
		if err != nil {
			return fmt.Errorf("abort existed %s dag failed: %s", dag.PlanCreateClassicRestore, err.Error())
		}

		tilog.Info(ctx, fmt.Sprintf("abort existed %s dag success", dag.PlanCreateClassicRestore),
			zap.String("restoreId", input.RestoreId),
			zap.Any("dag ins", utils.JSON(creationDagIns)))
	}

	// query existed delete dag ins
	deletionDagIns, err := s.dagInstanceFinder.GetRunningDagInstance(ctx, dag.PlanDeleteClassicRestore, filters...)
	if err != nil && !errors.Is(err, executorext.DagInsNotFoundErr) {
		tilog.Error(ctx, fmt.Sprintf("failed to get running %s dag", dag.PlanDeleteClassicRestore),
			zap.Error(err),
			zap.String("restoreId", input.RestoreId))
		return err
	}
	if deletionDagIns != nil {
		tilog.Info(ctx, fmt.Sprintf("there is a running %s dag, no need to create new one", dag.PlanDeleteClassicRestore),
			zap.String("restoreId", input.RestoreId),
			zap.Any("dag ins", utils.JSON(deletionDagIns)))
		return nil
	}

	// create deletion dag
	tilog.Info(ctx, fmt.Sprintf("create a %s dag ins ...", dag.PlanDeleteClassicRestore),
		zap.Any("filters", filters))
	args := actions.ActionArgs{
		utils.TraceIdKey:         utils.GetTraceId(ctx),
		workflow.RestoreIdKey:    actions.ActionArgString(input.RestoreId),
		workflow.SharedPoolIdKey: actions.ActionArgString(restore.TargetSharedPoolID),
	}
	err = s.dagExecutor.Submit(ctx, dag.PlanDeleteClassicRestore, args, fmt.Sprintf("del-cl-restore-%s", input.RestoreId))
	if err != nil {
		return fmt.Errorf("failed to submit %s dag: %s", dag.PlanDeleteClassicRestore, err.Error())
	}
	tilog.Info(ctx, fmt.Sprintf("submit %s dag success", dag.PlanDeleteClassicRestore),
		zap.String("restoreId", input.RestoreId))
	return nil
}