package cluster

import (
	"context"
	"fmt"
	"strconv"
	"strings"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	configbackuprepo "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/configbackup"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/tidbgroup"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

func (s *serviceImpl) Delete(
	ctx context.Context,
	req *globaldedicatedv1.DeleteClusterRequest,
) (*entity.Cluster, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Domain.DeleteCluster")
	defer span.End()

	var cluster *entity.Cluster

	err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		var err error
		cluster, err = s.clusterRepo.GetAndLoadComponents(ctx, req.ClusterId)
		if err != nil {
			return err
		}

		if cluster.State == globaldedicatedv1.Cluster_DELETED {
			return bizerr.New(bizerr.ErrCodeParams, "can't delete cluster. status=%s", cluster.State.String())
		}

		filter := &tidbgroup.Filter{
			Ids:        nil,
			ClusterIds: []uint64{req.ClusterId},
		}
		var tidbGroups []*entity.TidbNodeGroup
		tidbGroups, bErr := s.tidbGroupRepo.Query(ctx, filter)
		if bErr != nil {
			return bErr
		}
		var tidbGroupIDs []string
		var privateLinkServiceIDs []string
		for _, group := range tidbGroups {
			privateLinkServiceIDs = append(privateLinkServiceIDs, strconv.FormatUint(group.Id, 10))
			// default work pool will be deleted by regional delete cluster api
			if !group.Default {
				tidbGroupIDs = append(tidbGroupIDs, strconv.FormatUint(group.Id, 10))
			}
		}

		// set cluster state
		cluster.State = globaldedicatedv1.Cluster_DELETED
		cluster.HealthState = globaldedicatedv1.Cluster_UNHEALTHY

		if req.ValidateOnly {
			return nil
		}

		// Deleting event
		deletingEvent := entity.NewClusterEventBuilder(ctx, cluster, clusterevent.EventTypeDeleting).Build()
		if err := s.eventCommitter.Write(ctx, clusterevent.TopicClusterEvent, deletingEvent); err != nil {
			return err
		}

		// Delete all db records belonged to the cluster, which includes:
		// - Cluster itself
		// - Cluster component
		// - Worker pool
		// - Cluster cidr (ip allow list)
		// - Private link endpoint
		// - Cluster component snapshot
		// - (Other records related to the cluster added in the future)

		if err := s.clusterRepo.Save(ctx, cluster); err != nil {
			return err
		}

		if err := s.clusterRepo.Delete(ctx, cluster); err != nil {
			return err
		}

		if err := s.componentRepo.DeleteByClusterId(ctx, req.ClusterId); err != nil {
			return err
		}

		if err := s.tidbGroupRepo.DeleteByClusterId(ctx, req.ClusterId); err != nil {
			return err
		}

		if err := s.trafficFilterRepo.DeleteByClusterId(ctx, req.ClusterId); err != nil {
			return err
		}

		if err := s.privateLinkEndpointRepo.DeleteByClusterID(ctx, req.ClusterId); err != nil {
			return err
		}

		if err := s.pausePlanRepo.DeleteByClusterId(ctx, req.ClusterId); err != nil {
			return err
		}

		if err := s.configBackupPolicyRepo.Disable(ctx, req.ClusterId, &configbackuprepo.DisableOption{
			DeleteBackupPending: true,
			DeletePolicyPending: true,
		}); err != nil {
			return err
		}

		if err := s.auditLogConfigRepo.DeleteByClusterID(ctx, req.ClusterId); err != nil {
			return err
		}

		if err := s.auditLogFilterRuleRepo.DeleteByClusterID(ctx, req.ClusterId); err != nil {
			return err
		}

		if err := s.workflow.DeleteDagInstances(ctx, fmt.Sprintf("%d", cluster.ClusterId), map[string]bool{
			common.DeleteTidbGroup:             true,
			common.DeletePrivateLinkConnection: true,
			common.DeletePrivateLinkService:    true,
		}); err != nil {
			return err
		}

		// Start deleting workflow dag instance.
		workFlowErr := s.workflow.RunDagWithBizId(
			ctx,
			common.DeleteCluster,
			strconv.FormatUint(cluster.OrganizationId, 10),
			strconv.FormatUint(cluster.ProjectId, 10),
			strconv.FormatUint(req.ClusterId, 10),
			map[string]string{
				workflow.KeyTraceId:             workflow.GetTraceId(ctx),
				common.KeyClusterId:             strconv.FormatUint(req.ClusterId, 10),
				common.KeyTidbGroupIds:          strings.Join(tidbGroupIDs, ","),
				common.KeyPrivateLinkServiceIds: strings.Join(privateLinkServiceIDs, ","),
			},
		)
		if workFlowErr != nil {
			return bizerr.WrapErrorWithCode(workFlowErr, bizerr.ErrCodeUnknown, "create async workflow failed")
		}

		return nil
	})
	if err != nil {
		return nil, bizerr.WrapError(err, "delete cluster failed")
	}

	return cluster, nil
}