package cluster

import (
	"context"
	"errors"
	"fmt"
	"strconv"

	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"gorm.io/gorm"

	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/constant"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

func (svc *service) DeleteCluster(
	ctx context.Context,
	clusterId uint64,
	validateOnly bool,
) (*entity.Cluster, *entity.ClusterCustomStatus, error) {
	return svc.deleteCluster(ctx, clusterId, validateOnly, deleteOptions{})
}

type deleteOptions struct {
	needFinalBackup           bool
	deleteResourcePoolIfEmpty bool
}

func (svc *service) deleteCluster(
	ctx context.Context,
	clusterId uint64,
	validateOnly bool,
	options deleteOptions,
) (*entity.Cluster, *entity.ClusterCustomStatus, error) {
	spanName := "ClusterService.DeleteCluster"
	ctx, span := tracer.Start(ctx, spanName)
	defer span.End()

	var (
		cluster *entity.Cluster
		status  *entity.ClusterCustomStatus
	)

	if clusterId == 0 {
		return nil, nil, bizerr.New(bizerr.ErrCodeParams, "cluster id is invalid")
	}
	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		var (
			err     error
			isMutex bool
		)
		// First make sure cluster exists in database
		cluster, err = svc.clusterRepo.GetCluster(ctx, clusterId, false)
		if err != nil {
			return err
		}

		if cluster.DeployType != entity.DeployTypeNextGen {
			return fmt.Errorf("cluster deploy type is not nextgen, got: %s", cluster.DeployType)
		}

		// check ClusterAction_CLUSTER_ACTION_DELETE_CLUSTER mutex
		isMutex, status, err = svc.customStatusRepo.IsMutexClusterAction(
			ctx,
			clusterId,
			tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_DELETE_CLUSTER,
		)
		if err != nil {
			if !errors.Is(err, gorm.ErrRecordNotFound) {
				return fmt.Errorf("failed to check mutex custom status: %w", err)
			}
		}

		if isMutex {
			tilog.Error(ctx, "mutex custom status is exist", zap.Uint64("cluster_id", clusterId), zap.Any("custom status", status))
			return constant.ErrClusterCustomStatusForbidden
		}

		request := &globalngv1.DeleteClusterRequest{ClusterId: clusterId}
		request.ValidateOnly = true
		nextgenCluster, err := svc.nextGenCli.DeleteCluster(ctx, request)
		if err != nil {
			return err
		}

		if nextgenCluster == nil {
			return errors.New("global cluster object should not be nil")
		}
		var resourcePoolID uint64
		if options.deleteResourcePoolIfEmpty {
			resourcePoolID, err = strconv.ParseUint(nextgenCluster.GetSharedPoolId(), 10, 64)
			if err != nil {
				return fmt.Errorf("failed to parse resource pool id %q: %w", nextgenCluster.GetSharedPoolId(), err)
			}
			if resourcePoolID == 0 {
				return errors.New("resource pool id is invalid")
			}
		}

		if validateOnly {
			cluster = cvtNextGenCluster(cluster, nextgenCluster)
			return nil
		}

		// Delete cluster record in schema.
		// if validateOnly is false, we need to
		// 1. delete cluster record in database
		// 2. delete custom status record in schema
		// 3. delete maintenance task execution
		// 4. run delete cluster dag task
		// 5. remove cluster permissions

		if err := svc.clusterRepo.DeleteCluster(ctx, clusterId); err != nil {
			return fmt.Errorf("failed to delete cluster record: %w", err)
		}

		// Delete custom status record in schema.
		if err := svc.customStatusRepo.DeleteClusterCustomStatusByCluster(ctx, clusterId); err != nil {
			return fmt.Errorf("failed to delete custom status record in schema: %w", err)
		}

		if err := svc.workflow.DeleteDagInstances(ctx, fmt.Sprintf("%d", clusterId), map[string]bool{}); err != nil {
			return fmt.Errorf("failed to delete dag instances: %w", err)
		}

		// run dag instance
		request.ValidateOnly = false
		requestJson, err := protojson.Marshal(request)
		if err != nil {
			return fmt.Errorf("failed to marshal delete cluster request: %w", err)
		}
		// metadata
		metadataJson, err := metadata.MarshalRequestMetaFromCtx(ctx)
		if err != nil {
			return fmt.Errorf("failed to marshal request metadata: %w", err)
		}
		dagVars := map[string]string{
			constant.KeyRequest:     string(requestJson),
			workflow.KeyTraceId:     workflow.GetTraceId(ctx),
			workflow.KeySpanName:    spanName,
			workflow.KeyRequestMeta: string(metadataJson),
		}
		if options.needFinalBackup {
			dagVars[constant.KeyFinalBackup] = "true"
		}
		if options.deleteResourcePoolIfEmpty {
			dagVars[constant.KeyResourcePoolID] = strconv.FormatUint(resourcePoolID, 10)
			dagVars[constant.KeyOrganizationID] = strconv.FormatUint(cluster.OrganizationId, 10)
			dagVars[constant.KeyServicePlan] = strconv.FormatInt(int64(nextgenCluster.GetServicePlan()), 10)
			dagVars[constant.KeyDeleteResourcePoolIfEmpty] = "true"
		}
		err = svc.workflow.RunDagWithBizId(
			ctx,
			constant.DagNextGenDeleteCluster,
			strconv.FormatUint(cluster.OrganizationId, 10),
			"0",
			strconv.FormatUint(cluster.ClusterId, 10),
			dagVars,
		)
		if err != nil {
			return fmt.Errorf("failed to run dag for delete cluster: %w", err)
		}

		if err = svc.accountCli.DeleteClusterRelations(ctx, clusterId); err != nil {
			return fmt.Errorf("failed to delete cluster permissions: %w", err)
		}

		cluster = cvtNextGenCluster(cluster, nextgenCluster)
		return nil
	}); err != nil {
		return nil, status, fmt.Errorf("failed to delete cluster: %w", err)
	}

	return cluster, nil, nil
}