package server

import (
	"context"
	"fmt"
	"strings"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
)

func (s *serverImpl) DeleteTiDBCluster(ctx context.Context, tidbClusterId string) (*entity.TiDBCluster, error) {
	if tidbClusterId == "" {
		return nil, status.Errorf(codes.InvalidArgument, "tidbClusterId is empty")
	}

	var tidbCluster *entity.TiDBCluster
	var dagExecId string
	var dagSubmitted bool
	if err := s.transactioner.Transaction(ctx, func(ctx context.Context) error {
		var err error
		tidbCluster, err = s.tidbClusterRepo.Load(ctx, tidbClusterId, false)
		if err != nil {
			return err
		}

		tidbGroups, err := s.tidbGroupRepo.List(ctx, nil, nil, []string{tidbClusterId}, nil, false)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to list tidb groups: %s", err)
		}
		if len(tidbGroups) > 0 {
			return status.Errorf(codes.FailedPrecondition, "tidbCluster %s still has %d associated tidb groups, please delete them first", tidbClusterId, len(tidbGroups))
		}

		// query deleting tidb groups
		deleteGroupIns, err := s.getRunningDagIns(ctx,
			[]string{dag.PlanDeleteTiDBGroup},
			[]string{tidbCluster.GetSharedPoolId()},
			nil,
			[]string{tidbClusterId},
			nil)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to get running delete tidb group dag instance: %s", err)
		}
		if len(deleteGroupIns) > 0 {
			tilog.Info(ctx, "tidb cluster has deleting tidb groups, please wait for them to be deleted first", zap.String("tidbClusterId", tidbClusterId), zap.Int("deletingTidbGroupCount", len(deleteGroupIns)))
			return status.Errorf(codes.FailedPrecondition, "tidb cluster has deleting tidb groups, please wait for them to be deleted first")
		}

		sharedPool, err := s.sharedPoolRepo.Load(ctx, tidbCluster.GetSharedPoolId(), false)
		if err != nil {
			if status.Code(err) != codes.NotFound {
				return status.Errorf(codes.Internal, "failed to load sharedPool: %s", err)
			}
			// not found, check if it's in creating
			return status.Errorf(codes.NotFound, "sharedPool %s not found", tidbCluster.GetSharedPoolId())
		}
		if err := guardSharedPoolMutation(sharedPool, sharedPoolMutationDeleteInstance); err != nil {
			return err
		}

		// merge sharedPool status from dag task
		err = s.mergeSharedPoolStatusFromDagTask(ctx, sharedPool)
		if err != nil {
			tilog.Error(ctx, "failed to merge sharedPool status from dag task", zap.Error(err))
			return status.Errorf(codes.Internal, "failed to merge sharedPool status from dag task: %s", err)
		}

		// delete api not need to check sharedPool phase
		if sharedPool.Phase != apiv1.SharedPool_AVAILABLE {
			tilog.Warn(ctx, "sharedpool is not ready", zap.String("sharedPoolId", tidbCluster.GetSharedPoolId()), zap.String("status", sharedPool.GetPhase().String()))
		}

		// cancel cluster dag
		needCancelPlans := []string{
			dag.PlanCreateTiDBCluster,
			dag.PlanUpdateTiDBCluster,
		}

		dagIns, err := s.getRunningDagIns(ctx, needCancelPlans, []string{sharedPool.GetId()}, nil, []string{tidbClusterId}, nil)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to get running dag instances: %s", err)
		}
		// PlanCreateCassandraProxy stores the cluster ID in both TiDBClusterIdKey and
		// ComponentIdsKey, so both filters must use tidbClusterId here.
		cassandraDagIns, err := s.getRunningDagIns(ctx, []string{dag.PlanCreateCassandraProxy}, []string{sharedPool.GetId()}, nil, []string{tidbClusterId}, []string{tidbClusterId})
		if err != nil {
			return status.Errorf(codes.Internal, "failed to get running cassandra-proxy dag instances: %s", err)
		}
		for id, task := range cassandraDagIns {
			dagIns[id] = task
		}

		for id, task := range dagIns {
			// abort create cluster dag
			tilog.Info(ctx, "aborting cluster dag", zap.String("exec id", id), zap.String("root name", task.ExecTask.RootName))
			err = s.dagExecutor.Abort(ctx, task.ExecTask.ExecId)
			if err != nil {
				tilog.Error(ctx, "failed to abort cluster dag", zap.String("task id", task.ExecTask.ExecId), zap.String("root name", task.ExecTask.RootName), zap.Error(err))
				return status.Errorf(codes.Internal, "failed to abort cluster dag %s: %s", task.ExecTask.ExecId, err)
			}
		}

		var tiproxyComp *entity.Component
		if !entity.HasSharedTiProxy(sharedPool) {
			tiproxyComp = entity.FindTiProxyComponent(sharedPool, tidbCluster.GetId())
			if tiproxyComp == nil {
				tiproxyComp = &entity.Component{
					Id:            tidbCluster.GetId(),
					Phase:         apiv1.Component_DELETING,
					ComponentType: apiv1.ComponentType_TIPROXY,
					NodeQuota:     &entity.NodeQuota{NodeCount: 0},
				}
				sharedPool.ComponentGroups = append(sharedPool.ComponentGroups, tiproxyComp)
			}

			tiproxyComp.Phase = apiv1.Component_DELETING
			if tiproxyComp.NodeQuota != nil {
				tiproxyComp.GetNodeQuota().NodeCount = 0
			}
		}

		var cassandraProxyComp *entity.Component
		cassandraProxyState, err := s.cassandraProxyStateRepo.LoadByClusterID(ctx, tidbClusterId, false)
		if err != nil && status.Code(err) != codes.NotFound {
			return status.Errorf(codes.Internal, "failed to load cassandra-proxy state: %s", err)
		}
		if cassandraProxyState != nil {
			if err := s.cassandraProxyStateRepo.UpdatePhase(ctx, tidbClusterId, cassandraProxyState.HostID, apiv1.Component_DELETING); err != nil {
				return status.Errorf(codes.Internal, "failed to mark cassandra-proxy state deleting: %s", err)
			}
			sharedPool.ComponentGroups = removeComponentByIDAndType(
				sharedPool.ComponentGroups,
				tidbCluster.GetId(),
				apiv1.ComponentType_CASSANDRA_PROXY,
			)
			cassandraProxyComp = &entity.Component{
				Id:            tidbCluster.GetId(),
				Phase:         apiv1.Component_DELETING,
				ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
				NodeQuota:     cassandraProxyState.NodeQuotaEntity,
				Labels: map[string]string{
					cassandraProxyHostIdLabelKey:   cassandraProxyState.HostID,
					cassandraProxyNodeSpecLabelKey: cassandraProxyState.NodeSpec,
				},
			}
			sharedPool.ComponentGroups = append(sharedPool.ComponentGroups, cassandraProxyComp)
		}

		tidbCluster.Phase = apiv1.Cluster_UNAVAILABLE

		deleteIns, err := s.getRunningDagIns(ctx,
			[]string{dag.PlanDeleteTiDBCluster},
			[]string{sharedPool.GetId()},
			nil,
			[]string{tidbClusterId},
			nil)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to get running delete dag instance: %s", err)
		}

		if len(deleteIns) > 0 {
			tilog.Info(ctx, "delete tidb cluster dag is already running", zap.String("sharedPoolId", sharedPool.GetId()), zap.String("tidbClusterId", tidbClusterId))
			return nil
		}

		err = s.tidbClusterRepo.Delete(ctx, tidbClusterId)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to delete tidbCluster meta: %s", err)
		}

		// if shared pool is creating, just return, not need submit delete dag
		if sharedPool.Phase == apiv1.SharedPool_CREATING {
			return nil
		}

		sharedPoolStr, err := protojson.Marshal(sharedPool)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to marshal sharedPool: %s", err)
		}

		componentIds := make([]string, 0, 2)
		componentTypes := make([]string, 0, 2)
		if tiproxyComp != nil {
			componentIds = append(componentIds, tiproxyComp.GetId())
			componentTypes = append(componentTypes, entity.GetComponentTypeName(tiproxyComp))
		}
		if cassandraProxyComp != nil {
			componentIds = append(componentIds, cassandraProxyComp.GetId())
			componentTypes = append(componentTypes, entity.GetComponentTypeName(cassandraProxyComp))
		}

		args := actions.ActionArgs{
			workflow.SharedPoolIdKey:              actions.ActionArgString(sharedPool.GetId()),
			workflow.SharedPoolKey:                actions.ActionArgString(string(sharedPoolStr)),
			workflow.DataplaneIdKey:               actions.ActionArgString(sharedPool.GetDataplaneName()),
			workflow.ComponentIdsKey:              actions.ActionArgString(strings.Join(componentIds, ",")),
			workflow.ComponentTypesKey:            actions.ActionArgString(strings.Join(componentTypes, ",")),
			workflow.TiDBClusterIdKey:             actions.ActionArgString(tidbCluster.GetId()),
			workflow.MeteringTypeKey:              actions.ActionArgString(workflow.MeteringTypeValueLogicMetaByTiDBCluster),
			workflow.MeteringDeleteTiDBClusterKey: actions.ActionArgString(workflow.MeteringDeleteTiDBClusterValue),
			utils.TraceIdKey:                      utils.GetTraceId(ctx),
		}

		dagExecId = utils.GeneratedDagExecId(ctx, fmt.Sprintf("delete-tidbcluster-%s-%s", sharedPool.GetId(), tidbCluster.GetId()))
		err = s.dagExecutor.Submit(ctx, dag.PlanDeleteTiDBCluster, args, dagExecId)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to submit delete tidb cluster dag: %s", err)
		}
		dagSubmitted = true
		return nil
	}); err != nil {
		if dagSubmitted {
			if abortErr := s.dagExecutor.Abort(ctx, dagExecId); abortErr != nil {
				tilog.Error(ctx, "failed to abort dag after transaction failure",
					zap.String("dagExecId", dagExecId),
					zap.Error(abortErr))
			}
		}
		return nil, err
	}

	return tidbCluster, nil
}

func removeComponentByIDAndType(components []*entity.Component, id string, componentType apiv1.ComponentType) []*entity.Component {
	filtered := components[:0]
	for _, comp := range components {
		if comp.GetId() == id && comp.GetComponentType() == componentType {
			continue
		}
		filtered = append(filtered, comp)
	}
	return filtered
}