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"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/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,
	needFinalBackup, validateOnly bool,
) (*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, errors.New("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 cluster in database: %w", err)
		}

		if cluster.DeployType != entity.DeployTypeDedicated {
			return errors.New("cluster is not a dedicated cluster")
		}

		// 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 constant.ErrClusterCustomStatusForbidden
		}

		request := &globaldedicatedv1.DeleteClusterRequest{ClusterId: clusterId}
		request.ValidateOnly = true
		dedicatedCluster, err := svc.dedicatedCli.DeleteCluster(ctx, request)
		if err != nil {
			return err
		}

		if dedicatedCluster == nil {
			return errors.New("global cluster object should not be nil")
		}

		if validateOnly {
			cluster = cvtDedicatedCluster(cluster, dedicatedCluster)
			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

		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 custom status record in database: %w", err)
		}

		if err := svc.cancelMaintenanceTaskExecution(ctx, cluster); err != nil {
			return fmt.Errorf("failed to cancel maintenance task execution: %w", err)
		}

		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)
		}

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

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

		cluster = cvtDedicatedCluster(cluster, dedicatedCluster)
		return nil
	}); err != nil {
		return cluster, status, err
	}

	return cluster, nil, nil
}

func (svc *service) cancelMaintenanceTaskExecution(ctx context.Context, cluster *entity.Cluster) error {
	filter := &tidbmgmtservicev1.QueryMaintenanceTaskExecutionsRequest_Filter{
		ClusterIds: []uint64{cluster.ClusterId},
		States: []tidbmgmtservicev1.MaintenanceTaskExecution_State{
			tidbmgmtservicev1.MaintenanceTaskExecution_PENDING,
			tidbmgmtservicev1.MaintenanceTaskExecution_RUNNING,
		},
	}
	executions, err := svc.mteRepo.Query(ctx, filter, nil)
	if err != nil {
		return err
	}

	for _, execution := range executions {
		switch execution.State {
		case tidbmgmtservicev1.MaintenanceTaskExecution_PENDING:
			// continue
		case tidbmgmtservicev1.MaintenanceTaskExecution_RUNNING:
			// continue
		case tidbmgmtservicev1.MaintenanceTaskExecution_DONE:
			return bizerr.New(bizerr.ErrCodeParams, "cannot cancel a finished task execution")
		case tidbmgmtservicev1.MaintenanceTaskExecution_CANCELED:
			return bizerr.New(bizerr.ErrCodeParams, "task execution is already canceled")
		}

		execution.State = tidbmgmtservicev1.MaintenanceTaskExecution_CANCELED
		if cluster.DeletedAt != nil {
			execution.CancellationReason = tidbmgmtservicev1.MaintenanceTaskExecution_CLUSTER_DELETED
		} else {
			execution.CancellationReason = tidbmgmtservicev1.MaintenanceTaskExecution_OPS_CANCELED
		}

		if err := svc.mteRepo.Save(ctx, execution); err != nil {
			return err
		}
	}

	return nil
}