package cluster

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

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

	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/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/tidb-mgmt-service/domain/service/serverless"
	"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,
) (*entity.Cluster, *entity.ClusterCustomStatus, error) {
	spanName := "ClusterService.DeleteCluster"
	ctx, span := serverless.Tracer.Start(ctx, spanName)
	defer span.End()

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

	if clusterId == 0 {
		return nil, nil, fmt.Errorf("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 fmt.Errorf("failed to find cluste in database: %w", err)
		}

		if cluster.DeployType != entity.DeployTypeServerless {
			return errtrace.Wrap(errors.New("cluster deploy type is not serverless"))
		}

		// 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 cluster action: %w", err)
			}
		}

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

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

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

		// delete maintenance task execution
		if err := svc.workflow.DeleteDagInstances(ctx, fmt.Sprintf("%d", clusterId), map[string]bool{}); err != nil {
			return fmt.Errorf("failed to cancel relation tasks: %w", err)
		}

		// get cluster from global serverless
		clusters, err := svc.serverlessCli.ListClusters(ctx, &globalserverlessv1.ServerlessServiceListRequest{
			Filter: &globalserverlessv1.ClusterFilter{ClusterIds: []string{strconv.FormatUint(clusterId, 10)}},
		})
		if err != nil {
			return bizerr.WrapGormError(err, "failed to get cluster from global serverless")
		}

		if len(clusters.GetItems()) == 0 {
			tilog.Warn(ctx, "cluster not found in global serverless", zap.Uint64("cluster_id", clusterId))
			cluster = cvtServerlessCluster(cluster, defaultServerlessCluster(ctx, cluster))
			return nil
		}

		// check cluster can be deleted
		// Fall back to project_id as original_project_id for legacy clusters where
		// original_project_id is 0 in the DB (supplement migration not yet run).
		originalProjectId := cluster.OriginalProjectId
		if originalProjectId == 0 {
			originalProjectId = cluster.ProjectId
		}
		request := &globalserverlessv1.ServerlessServiceDeleteRequest{
			ClusterId:         strconv.FormatUint(clusterId, 10),
			ProjectId:         cluster.ProjectId,
			OriginalProjectId: originalProjectId,
		}
		request.ValidateOnly = true
		resp, err := svc.serverlessCli.DeleteCluster(ctx, request)
		if err != nil {
			return errtrace.Wrap(err)
		}
		if resp.Cluster == nil {
			return errtrace.Wrap(errors.New("regional cluster object should not be nil"))
		}

		// run delete cluster dag task
		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)
		}
		err = svc.workflow.RunDagWithBizId(
			ctx,
			constant.DagServerlessDeleteCluster,
			strconv.FormatUint(cluster.OrganizationId, 10),
			strconv.FormatUint(cluster.ProjectId, 10),
			strconv.FormatUint(cluster.ClusterId, 10),
			map[string]string{
				constant.KeyRequest:     string(requestJson),
				workflow.KeyTraceId:     workflow.GetTraceId(ctx),
				workflow.KeySpanName:    spanName,
				workflow.KeyRequestMeta: string(metadataJson),
			},
		)
		if err != nil {
			return fmt.Errorf("failed to run delete serverless cluster dag task: %w", err)
		}

		if err = svc.accountCli.DeleteClusterRelations(ctx, cluster.ClusterId); err != nil {
			tilog.Error(ctx, "failed to delete cluster relations to account-svc", zap.Error(err))
		}

		cluster = cvtServerlessCluster(cluster, resp.Cluster)

		return nil
	}); err != nil {
		return cluster, status, errtrace.Wrap(err)
	}

	return cluster, nil, nil
}