package cluster

import (
	"context"
	"fmt"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/cvt"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

func (svc *clusterServiceImpl) Delete(
	ctx context.Context,
	id string,
	validateOnly bool,
) (*globalserverlessv1.ServerlessServiceCluster, error) {
	ctx, span := Tracer.Start(ctx, "Domain.Delete")
	defer span.End()

	// Load cluster from repo.
	cluster, err := svc.clusterRepo.Get(ctx, id, false)
	if err != nil {
		return nil, fmt.Errorf("load cluster failed: %w", err)
	}

	if cluster == nil {
		return nil, fmt.Errorf("cluster is not found: %w", err)
	}

	if !cluster.DeletedAt.IsZero() {
		return nil, fmt.Errorf("cluster is already deleted: %w", err)
	}
	if regionalClusters, _ := svc.regionalService.ListCluster(
		ctx,
		cluster.Region,
		&globalserverlessv1.ServerlessServiceListRequest{
			Filter: &globalserverlessv1.ClusterFilter{
				ClusterIds: []string{cluster.ClusterId},
				ProjectIds: []uint64{cluster.ProjectId},
			},
		},
	); len(regionalClusters) == 1 {
		cluster.ServicePlan = globalcommonv1.ServicePlan_Starter
		if regionalClusters[0].GetAutoScaling().GetMaxRcu() > 0 {
			cluster.ServicePlan = globalcommonv1.ServicePlan_Essential
		}
	}

	// Build output cluster.
	pb := cvt.ClusterEnToGlobal(cluster)

	// Force global status to deleted and regional status to deleting.
	cluster.UserStatus = globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_DELETED
	cluster.Cluster.State = regionalserverlessv1.ClusterState_CLUSTER_STATE_DELETING

	// Early return if validate only.
	if validateOnly {
		return pb, nil
	}

	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		if err := svc.clusterRepo.Delete(ctx, id); err != nil {
			return err
		}

		evs, err := cluster.DeleteEvent(ctx)
		if err != nil {
			return err
		}

		if err := svc.eventCommitter.BatchWrite(ctx, clusterevent.TopicClusterEvent, evs); err != nil {
			return err
		}

		// delete create/update workflow
		if err := svc.flow.DeleteDagInstances(ctx, cluster.ClusterId, map[string]bool{
			common.DeleteCluster: true,
		}); err != nil {
			return err
		}

		if err := svc.flow.RunDagWithBizId(
			ctx,
			common.DeleteCluster,
			fmt.Sprintf("%d", cluster.TenantId),
			fmt.Sprintf("%d", cluster.ProjectId),
			cluster.ClusterId,
			map[string]string{
				workflow.KeyTraceId: workflow.GetTraceId(ctx),
				common.KeyClusterId: cluster.ClusterId,
			},
		); err != nil {
			return err
		}

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

	return pb, nil
}