package cluster

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

	"github.com/samber/lo"

	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	exeportrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/export"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/privatelink"
	tidbgrouprepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/tidbgroup"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/resourcepoolpolicy"
	servicecommon "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/common"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/schema"
	"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 *globalngv1.DeleteClusterRequest,
) (*entity.Cluster, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Domain.DeleteCluster")
	defer span.End()

	var cluster *entity.Cluster
	cluster, err := s.clusterRepo.Get(ctx, req.ClusterId, false)
	if err != nil {
		return nil, err
	}
	if cluster == nil {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster not found")
	}

	if cluster.GetDeleteTime() != nil || cluster.State == globalngv1.Cluster_DELETED {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster is already deleted")
	}
	resourcePoolID, parseErr := strconv.ParseUint(cluster.GetSharedPoolId(), 10, 64)
	if parseErr != nil || resourcePoolID == 0 {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "cluster has an invalid resource pool id")
	}
	resourcePool, err := s.resourcePoolRepo.Load(ctx, resourcePoolID, false)
	if err != nil {
		return nil, err
	}
	if lifecycleErr := servicecommon.RequireResourcePoolLifecycle(
		resourcePool,
		resourcepoolpolicy.OperationDeleteInstance,
	); lifecycleErr != nil {
		return nil, lifecycleErr
	}

	exportRunningCount, err := s.exportRepo.Count(ctx, &exeportrepo.QueryExportsFilter{
		ClusterIds: []string{strconv.FormatUint(req.ClusterId, 10)},
		States:     []schema.ExportState{schema.ExportStateRunning},
	})
	if err != nil {
		return nil, bizerr.WrapError(err, "failed to count running exports")
	}
	if exportRunningCount > 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster has running exports, please cancel them first")
	}

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

	if req.ValidateOnly {
		return cluster, nil
	}

	if err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		lockedResourcePool, lockErr := s.resourcePoolRepo.LoadForUpdate(ctx, resourcePoolID, false)
		if lockErr != nil {
			return lockErr
		}
		if lifecycleErr := servicecommon.RequireResourcePoolLifecycle(
			lockedResourcePool,
			resourcepoolpolicy.OperationDeleteInstance,
		); lifecycleErr != nil {
			return lifecycleErr
		}
		tidbgroups, bErr := s.tidbgroupRepo.Query(ctx, &tidbgrouprepo.Filter{
			ClusterIds: []uint64{req.ClusterId},
		})
		if bErr != nil {
			return bErr
		}
		tidbgroupIds := lo.Map(tidbgroups, func(tg *entity.TidbGroup, _ int) string {
			return strconv.FormatUint(tg.Id, 10)
		})

		// Deleting event
		evs, err := cluster.DeleteEvent(ctx)
		if err != nil {
			return err
		}
		if err := s.eventCommitter.BatchWrite(ctx, clusterevent.TopicClusterEvent, evs); err != nil {
			return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "write cluster event failed")
		}

		services, err := s.privatelinkRepo.ListPrivateLinkServices(ctx, &privatelink.Filter{
			ClusterID: req.ClusterId,
		})
		if err != nil {
			return err
		}
		for _, service := range services {
			if err := s.privatelinkRepo.DeletePrivateLinkConnectionByServiceID(ctx, service.PrivateLinkServiceID); err != nil {
				return err
			}
			if err := s.privatelinkRepo.DeletePrivateLinkService(ctx, service.PrivateLinkServiceID); err != nil {
				return err
			}
		}

		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.tidbgroupRepo.DeleteByClusterId(ctx, req.ClusterId); err != nil {
			return err
		}

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

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

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

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

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

		if err := s.backupSettingRepo.Delete(ctx, req.ClusterId); err != nil && err.ErrCode() != bizerr.ErrCodeRecordNotFound {
			return err
		}

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

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

		// Start deleting workflow dag instance.
		workFlowErr := s.workflow.RunDagWithBizId(
			ctx,
			common.DeleteCluster,
			strconv.FormatUint(cluster.OrganizationId, 10),
			"0",
			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.KeyTidbGroupIdWithPrivateLinkServiceIds: common.GenerateTidbGroupIdWithPrivateLinkServiceIds(services),
			},
		)
		if workFlowErr != nil {
			return bizerr.WrapErrorWithCode(workFlowErr, bizerr.ErrCodeUnknown, "create async workflow failed")
		}

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

	return cluster, nil
}