package server

import (
	"context"
	"errors"
	"fmt"

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

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

	restoreAdaptor := br.GetLightweightRestoreAdaptor(restore, s.clustergatewayCli, s.sharedPoolRepo)
	deletionDagMeta := restoreAdaptor.GetDeletionDagMeta(req.RestoreId, req.TargetSharedPoolId)
	creationDagMeta := restoreAdaptor.GetCreationDagMeta(req.RestoreId, req.TargetSharedPoolId)

	// Cancel create dag if exist
	creationDagIns, err := s.dagInstanceFinder.GetRunningDagInstance(ctx, creationDagMeta.DagName, creationDagMeta.DagInstanceFilters...)
	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", creationDagMeta.DagName),
					zap.Error(ierr),
					zap.String("restoreId", req.RestoreId),
					zap.Any("dag ins", utils.JSON(creationDagIns)))
				return fmt.Errorf("abort existed %s dag failed: %w", creationDagMeta.DagName, ierr)
			}
			ierr = s.lwRestoreRepo.SetStatus(ctx, req.RestoreId, schema.LightweightRestoreStatusCancelled)
			if ierr != nil {
				tilog.Error(ctx, "mark db record as cancelled failed",
					zap.Error(ierr),
					zap.String("restoreId", req.RestoreId))
				return fmt.Errorf("mark db record as cancelled: %w", ierr)
			}
			return nil
		})
		if err != nil {
			return fmt.Errorf("abort existed %s dag failed: %s", creationDagMeta.DagName, err.Error())
		}

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

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

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