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 DeleteManualBackupInput struct {
	*apiv1.DeleteManualBackupRequest
}

func (s *serverImpl) DeleteManualBackup(ctx context.Context, input *DeleteManualBackupInput) error {
	backup, err := s.manualBackupRepo.Load(ctx, input.ClusterId, input.BackupId, true)
	if err != nil {
		return fmt.Errorf("failed to load manual backup: %w", err)
	}
	if backup.DeletedAt.Valid || backup.Status == schema.ManualBackupStatusDeleting || backup.Status == schema.ManualBackupStatusDeleted {
		tilog.Info(ctx, "manual backup is already deleted or deleting, no need to create deletion dag", zap.String("backupId", input.BackupId))
		return nil
	}
	if _, err := s.guardSharedPoolCleanupMutationByID(ctx, backup.SharedPoolID); err != nil {
		return err
	}

	filters := []executorext.DagInsFilter{
		{
			ArgKey:    workflow.ManualBackupIdKey,
			ArgValues: []string{input.BackupId},
		},
		{
			ArgKey:    workflow.TiDBClusterIdKey,
			ArgValues: []string{backup.ClusterID},
		},
	}

	creationDagIns, err := s.dagInstanceFinder.GetRunningDagInstance(ctx, dag.PlanCreateManualBackup, filters...)
	if err != nil && !errors.Is(err, executorext.DagInsNotFoundErr) {
		return err
	}
	if creationDagIns != nil {
		err := s.dagExecutor.Abort(ctx, creationDagIns.ExecTask.ExecId)
		if err != nil {
			tilog.Error(ctx, fmt.Sprintf("abort existed %s dag failed", dag.PlanCreateManualBackup),
				zap.Error(err),
				zap.Any("filters", filters))
			return fmt.Errorf("abort existed %s dag failed: %w", dag.PlanCreateManualBackup, err)
		}
		tilog.Info(ctx, fmt.Sprintf("abort existed %s dag success", dag.PlanCreateManualBackup),
			zap.Any("filters", filters))
	}

	deletionDagIns, err := s.dagInstanceFinder.GetRunningDagInstance(ctx, dag.PlanDeleteManualBackup, filters...)
	if err != nil && !errors.Is(err, executorext.DagInsNotFoundErr) {
		tilog.Error(ctx, fmt.Sprintf("failed to get running %s dag", dag.PlanDeleteManualBackup),
			zap.Error(err),
			zap.Any("filters", filters))
		return err
	}
	if deletionDagIns != nil {
		tilog.Info(ctx, fmt.Sprintf("there is a running %s dag, no need to create new one", dag.PlanDeleteManualBackup),
			zap.Any("filters", filters))
		return nil
	}

	tilog.Info(ctx, fmt.Sprintf("create a %s dag ins ...", dag.PlanDeleteManualBackup), zap.Any("filters", filters))
	args := actions.ActionArgs{
		utils.TraceIdKey:           utils.GetTraceId(ctx),
		workflow.ManualBackupIdKey: actions.ActionArgString(backup.BackupID),
		workflow.TiDBClusterIdKey:  actions.ActionArgString(backup.ClusterID),
		workflow.SharedPoolIdKey:   actions.ActionArgString(backup.SharedPoolID),
	}
	if err := s.dagExecutor.Submit(ctx, dag.PlanDeleteManualBackup, args, manualBackupDeletionTaskId(&backup.ManualBackup)); err != nil {
		return fmt.Errorf("failed to submit %s dag: %s", dag.PlanDeleteManualBackup, err.Error())
	}
	tilog.Info(ctx, fmt.Sprintf("submit %s dag success", dag.PlanDeleteManualBackup), zap.Any("filters", filters))

	return nil
}

func manualBackupDeletionTaskId(backup *schema.ManualBackup) string {
	return fmt.Sprintf("mb-del-%s-%s", backup.ClusterID, backup.BackupID)
}