package cluster

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

	ffentity "github.com/tidbcloud/fastflow/pkg/entity"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/gkeheterogeneous"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/workflow/dag"
	"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"
)

const (
	applyGKEHeterogeneousDagNamePrefix  = "apply_gke_heterogeneous_deploy"
	deleteGKEHeterogeneousDagNamePrefix = "delete_gke_heterogeneous_deploy"
)

func gkeHeterogeneousBizId(clusterID uint64, componentType globaldedicatedv1.ComponentType, tidbGroupID uint64) string {
	return fmt.Sprintf("%d_%s_%d", clusterID, componentType.String(), tidbGroupID)
}

func gkeHeterogeneousNamePrefix(dagNamePrefix, bizId string) string {
	return fmt.Sprintf("%s_%s_%s", workflow.DynamicDagPrefix, dagNamePrefix, bizId)
}

func (s *serviceImpl) ApplyGKEHeterogeneousDeploy(
	ctx context.Context,
	req *globaldedicatedv1.ApplyGKEHeterogeneousDeployRequest,
) (*globaldedicatedv1.GKEHeterogeneousDeploy, bizerr.BizError) {
	if req.GetComponentType() == globaldedicatedv1.ComponentType_TIDB && req.GetTidbGroupId() != 0 {
		_, berr := s.tidbGroupRepo.Load(ctx, req.GetClusterId(), req.GetTidbGroupId())
		if berr != nil {
			if berr.ErrCode() == bizerr.ErrCodeRecordNotFound {
				return nil, bizerr.New(
					bizerr.ErrCodeParams,
					"tidb_group_id %d does not exist for cluster %d",
					req.GetTidbGroupId(),
					req.GetClusterId(),
				)
			}
			return nil, bizerr.WrapError(berr, "load tidb group failed")
		}
	}
	out, berr := gkeheterogeneous.PreviewApply(
		ctx,
		s.clusterRepo,
		s.regionRepo,
		s.featGateCli,
		s.regionServerCli,
		req,
	)
	if berr != nil {
		return nil, berr
	}
	if berr := s.checkNoPendingHeterogeneousDeploy(ctx, req.GetClusterId(), req.GetComponentType(), req.GetTidbGroupId()); berr != nil {
		return nil, berr
	}
	requestBody, err := protojson.Marshal(req)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeJsonMarshal, "marshal gke heterogeneous apply request failed")
	}
	metadataJSON, merr := metadata.MarshalRequestMetaFromCtx(ctx)
	if merr != nil {
		return nil, bizerr.WrapErrorWithCode(merr, bizerr.ErrCodeUnknown, "marshal request metadata failed")
	}

	concurrencyKey := gkeHeterogeneousConcurrencyKey(
		req.GetClusterId(),
		req.GetComponentType(),
		req.GetTidbGroupId(),
		req.GetAlternativeInstanceType(),
	)
	bizId := gkeHeterogeneousBizId(req.GetClusterId(), req.GetComponentType(), req.GetTidbGroupId())
	if berr := s.checkNoRunningHeterogeneousDag(ctx, applyGKEHeterogeneousDagNamePrefix, bizId, concurrencyKey); berr != nil {
		return nil, berr
	}

	if err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		if err := s.workflow.RegisterDynamicDagThenRun(ctx, applyGKEHeterogeneousDagNamePrefix, &workflow.CreateDynamicDagOption{
			OrgId:     strconv.FormatUint(req.GetTenantId(), 10),
			ProjectId: strconv.FormatUint(req.GetProjectId(), 10),
			BizId:     bizId,
			DagVars: map[string]string{
				workflow.KeyTraceId:                      workflow.GetTraceId(ctx),
				workflow.KeyRequestMeta:                  string(metadataJSON),
				common.KeyGKEHeterogeneousApplyReq:       string(requestBody),
				common.KeyGKEHeterogeneousConcurrencyKey: concurrencyKey,
			},
			DagTasks: dag.ApplyGKEHeterogeneousDeployDag().Tasks,
		}); err != nil {
			return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "create async workflow failed")
		}
		return nil
	}); err != nil {
		return nil, bizerr.WrapError(err, "apply gke heterogeneous deploy failed")
	}
	return out, nil
}

func (s *serviceImpl) GetGKEHeterogeneousDeploy(
	ctx context.Context,
	req *globaldedicatedv1.GetGKEHeterogeneousDeployRequest,
) (*globaldedicatedv1.GKEHeterogeneousDeploy, bizerr.BizError) {
	if !gkeheterogeneous.IsSupportedRegionalHeterogeneousComponentType(req.GetComponentType()) {
		return &globaldedicatedv1.GKEHeterogeneousDeploy{
			ComponentType: req.GetComponentType(),
		}, nil
	}

	bizId := gkeHeterogeneousBizId(req.GetClusterId(), req.GetComponentType(), req.GetTidbGroupId())
	compType := req.GetComponentType()
	tidbGroup := req.GetTidbGroupId()

	regionalResult, rerr := gkeheterogeneous.Get(
		ctx,
		s.clusterRepo,
		s.regionRepo,
		s.regionServerCli,
		req,
	)
	if rerr != nil {
		tilog.Error(ctx, "get regional heterogeneous deploy failed, checking pending dag anyway",
			zap.Error(rerr))
		applyPending, berr := s.querySinglePendingHeterogeneousDeploy(
			ctx, applyGKEHeterogeneousDagNamePrefix, bizId, compType, tidbGroup)
		if berr != nil {
			return nil, rerr
		}
		if applyPending != nil {
			return applyPending, nil
		}
		deletePending, berr := s.querySinglePendingHeterogeneousDeploy(
			ctx, deleteGKEHeterogeneousDagNamePrefix, bizId, compType, tidbGroup)
		if berr != nil {
			return nil, rerr
		}
		if deletePending != nil {
			return deletePending, nil
		}
		return nil, rerr
	}

	if regionalResult.GetEnabled() {
		pending, berr := s.querySinglePendingHeterogeneousDeploy(
			ctx, deleteGKEHeterogeneousDagNamePrefix, bizId, compType, tidbGroup)
		if berr != nil {
			return nil, berr
		}
		if pending != nil {
			return pending, nil
		}
		return regionalResult, nil
	}

	pending, berr := s.querySinglePendingHeterogeneousDeploy(
		ctx, applyGKEHeterogeneousDagNamePrefix, bizId, compType, tidbGroup)
	if berr != nil {
		return nil, berr
	}
	if pending != nil {
		return pending, nil
	}
	return regionalResult, nil
}

func (s *serviceImpl) querySinglePendingHeterogeneousDeploy(
	ctx context.Context,
	dagNamePrefix string,
	bizId string,
	componentType globaldedicatedv1.ComponentType,
	tidbGroupID uint64,
) (*globaldedicatedv1.GKEHeterogeneousDeploy, bizerr.BizError) {
	namePrefix := gkeHeterogeneousNamePrefix(dagNamePrefix, bizId)
	tilog.Info(ctx, "query pending heterogeneous dag",
		zap.String("dag_prefix", dagNamePrefix),
		zap.String("biz_id", bizId),
		zap.String("name_prefix", namePrefix),
		zap.String("component_type", componentType.String()),
		zap.Uint64("tidb_group_id", tidbGroupID))

	instances, err := s.workflow.ListRunningDagInstancesByNamePrefix(
		ctx,
		namePrefix,
		bizId,
	)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "list running dag instances failed")
	}

	tilog.Info(ctx, "pending heterogeneous dag query result",
		zap.Int("total_instances", len(instances)))

	for _, di := range instances {
		tilog.Info(ctx, "pending heterogeneous dag candidate",
			zap.String("dag_id", di.DagID),
			zap.String("status", string(di.Status)),
			zap.String("dag_ins_id", di.ID))

		switch dagNamePrefix {
		case deleteGKEHeterogeneousDagNamePrefix:
			tilog.Info(ctx, "pending heterogeneous dag matched: DELETING",
				zap.String("dag_ins_id", di.ID))
			return &globaldedicatedv1.GKEHeterogeneousDeploy{
				Enabled:       true,
				ComponentType: componentType,
				TidbGroupId:   tidbGroupID,
				Status:        "DELETING",
			}, nil
		case applyGKEHeterogeneousDagNamePrefix:
			applyReq, berr := parseApplyReqFromDagVars(di.Vars)
			if berr != nil {
				tilog.Error(ctx, "pending heterogeneous dag parse apply request failed",
					zap.String("dag_ins_id", di.ID),
					zap.Error(berr))
				return nil, berr
			}
			tilog.Info(ctx, "pending heterogeneous dag matched: APPLYING",
				zap.String("dag_ins_id", di.ID),
				zap.String("current_instance_type", applyReq.GetCurrentInstanceType()),
				zap.String("alternative_instance_type", applyReq.GetAlternativeInstanceType()))
			return &globaldedicatedv1.GKEHeterogeneousDeploy{
				Enabled:                 true,
				ComponentType:           applyReq.GetComponentType(),
				CurrentInstanceType:     applyReq.GetCurrentInstanceType(),
				AlternativeInstanceType: applyReq.GetAlternativeInstanceType(),
				TidbGroupId:             applyReq.GetTidbGroupId(),
				Status:                  "APPLYING",
			}, nil
		}
	}
	tilog.Info(ctx, "no active pending heterogeneous dag found, falling back to regional result",
		zap.String("dag_prefix", dagNamePrefix),
		zap.String("biz_id", bizId),
		zap.Int("candidates_checked", len(instances)))
	return nil, nil //nolint:nilnil // nil signals no pending DAG found
}

func parseApplyReqFromDagVars(vars ffentity.DagInstanceVars) (*globaldedicatedv1.ApplyGKEHeterogeneousDeployRequest, bizerr.BizError) {
	applyReqVar, ok := vars[common.KeyGKEHeterogeneousApplyReq]
	if !ok || applyReqVar.Value == "" {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "apply request not found in dag vars")
	}
	req := &globaldedicatedv1.ApplyGKEHeterogeneousDeployRequest{}
	if err := protojson.Unmarshal([]byte(applyReqVar.Value), req); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "unmarshal apply request from dag vars failed")
	}
	return req, nil
}

func (s *serviceImpl) DeleteGKEHeterogeneousDeploy(
	ctx context.Context,
	req *globaldedicatedv1.DeleteGKEHeterogeneousDeployRequest,
) bizerr.BizError {
	if req.GetComponentType() == globaldedicatedv1.ComponentType_TIDB && req.GetTidbGroupId() == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "tidb_group_id is required for tidb heterogeneous deploy")
	}
	if req.GetComponentType() != globaldedicatedv1.ComponentType_TIDB && req.GetTidbGroupId() != 0 {
		return bizerr.New(bizerr.ErrCodeParams, "tidb_group_id must be 0 for non-tidb component")
	}
	if !gkeheterogeneous.IsSupportedRegionalHeterogeneousComponentType(req.GetComponentType()) {
		return bizerr.New(bizerr.ErrCodeParams,
			"unsupported component_type %s for gke heterogeneous delete", req.GetComponentType().String())
	}
	_, regionEn, berr := gkeheterogeneous.Prepare(
		ctx,
		s.clusterRepo,
		s.regionRepo,
		nil,
		req.GetTenantId(),
		req.GetProjectId(),
		req.GetClusterId(),
		req.GetComponentType(),
		true,
		false,
	)
	if berr != nil {
		return berr
	}
	regionalComponent, cErr := gkeheterogeneous.ToRegionalComponentTypeOrError(req.GetComponentType())
	if cErr != nil {
		return cErr
	}

	if berr := s.checkNoPendingHeterogeneousDeploy(ctx, req.GetClusterId(), req.GetComponentType(), req.GetTidbGroupId()); berr != nil {
		return berr
	}

	clusterIDStr := strconv.FormatUint(req.GetClusterId(), 10)
	tidbGroupIDStr := strconv.FormatUint(req.GetTidbGroupId(), 10)
	if req.GetTidbGroupId() == 0 {
		tidbGroupIDStr = ""
	}
	deploy, deployExists, dErr := gkeheterogeneous.GetDeploy(
		ctx,
		s.regionServerCli,
		regionEn,
		clusterIDStr,
		regionalComponent,
		tidbGroupIDStr,
	)
	if dErr != nil {
		return bizerr.WrapErrorWithCode(dErr, bizerr.ErrCodeUnknown, "get gke heterogeneous deploy failed")
	}
	if !deployExists || len(deploy.GetInstanceTypes()) < 2 {
		return bizerr.New(bizerr.ErrCodeRecordNotFound,
			"gke heterogeneous deploy does not exist for cluster %d", req.GetClusterId())
	}

	concurrencyKey := gkeHeterogeneousConcurrencyKey(
		req.GetClusterId(),
		req.GetComponentType(),
		req.GetTidbGroupId(),
		deploy.GetInstanceTypes()[len(deploy.GetInstanceTypes())-1],
	)

	bizId := gkeHeterogeneousBizId(req.GetClusterId(), req.GetComponentType(), req.GetTidbGroupId())
	if berr := s.checkNoRunningHeterogeneousDag(ctx, deleteGKEHeterogeneousDagNamePrefix, bizId, concurrencyKey); berr != nil {
		return berr
	}

	requestBody, err := protojson.Marshal(req)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeJsonMarshal, "marshal gke heterogeneous delete request failed")
	}
	metadataJSON, merr := metadata.MarshalRequestMetaFromCtx(ctx)
	if merr != nil {
		return bizerr.WrapErrorWithCode(merr, bizerr.ErrCodeUnknown, "marshal request metadata failed")
	}

	if err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		if err := s.workflow.RegisterDynamicDagThenRun(ctx, deleteGKEHeterogeneousDagNamePrefix, &workflow.CreateDynamicDagOption{
			OrgId:     strconv.FormatUint(req.GetTenantId(), 10),
			ProjectId: strconv.FormatUint(req.GetProjectId(), 10),
			BizId:     bizId,
			DagVars: map[string]string{
				workflow.KeyTraceId:                      workflow.GetTraceId(ctx),
				workflow.KeyRequestMeta:                  string(metadataJSON),
				common.KeyGKEHeterogeneousDeleteReq:      string(requestBody),
				common.KeyGKEHeterogeneousConcurrencyKey: concurrencyKey,
			},
			DagTasks: dag.DeleteGKEHeterogeneousDeployDag().Tasks,
		}); err != nil {
			return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "create async workflow failed")
		}
		return nil
	}); err != nil {
		return bizerr.WrapError(err, "delete gke heterogeneous deploy failed")
	}
	return nil
}

func (s *serviceImpl) checkNoPendingHeterogeneousDeploy(
	ctx context.Context,
	clusterID uint64,
	componentType globaldedicatedv1.ComponentType,
	tidbGroupID uint64,
) bizerr.BizError {
	concurrencyPrefix := fmt.Sprintf("%d_%s_", clusterID, componentType.String())
	if tidbGroupID != 0 {
		concurrencyPrefix = fmt.Sprintf("%d_%s_%d_", clusterID, componentType.String(), tidbGroupID)
	}
	bizId := gkeHeterogeneousBizId(clusterID, componentType, tidbGroupID)

	applyInstances, err := s.workflow.ListRunningDagInstancesByNamePrefix(
		ctx,
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "list running apply dag instances failed")
	}
	deleteInstances, err := s.workflow.ListRunningDagInstancesByNamePrefix(
		ctx,
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "list running delete dag instances failed")
	}

	allInstances := make([]*ffentity.DagInstance, 0, len(applyInstances)+len(deleteInstances))
	allInstances = append(allInstances, applyInstances...)
	allInstances = append(allInstances, deleteInstances...)
	for _, di := range allInstances {
		if di.Status == ffentity.DagInstanceStatusSuccess {
			continue
		}
		key := diConcurrencyKey(di)
		if key != "" && strings.HasPrefix(key, concurrencyPrefix) {
			return bizerr.New(bizerr.ErrCodeParams,
				"a heterogeneous deploy operation is already in progress for this component. "+
					"Delete or wait for the existing operation to finish before applying another fallback instance type")
		}
	}
	return nil
}

func (s *serviceImpl) checkNoRunningHeterogeneousDag(
	ctx context.Context,
	dagNamePrefix string,
	bizId string,
	concurrencyKey string,
) bizerr.BizError {
	runningDags, err := s.workflow.ListRunningDagInstancesByNamePrefix(
		ctx,
		gkeHeterogeneousNamePrefix(dagNamePrefix, bizId),
		bizId,
	)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "check running dag instances failed")
	}

	for _, di := range runningDags {
		if di.Status == ffentity.DagInstanceStatusSuccess {
			continue
		}
		key := diConcurrencyKey(di)
		if key == concurrencyKey {
			return bizerr.New(bizerr.ErrCodeParams,
				"a gke heterogeneous deploy operation is already in progress for %s", concurrencyKey)
		}
	}
	return nil
}

func (s *serviceImpl) checkNoApplyHeterogeneousDeployPending(
	ctx context.Context,
	clusterID uint64,
	componentType globaldedicatedv1.ComponentType,
	tidbGroupID uint64,
) bizerr.BizError {
	bizId := gkeHeterogeneousBizId(clusterID, componentType, tidbGroupID)
	applyInstances, err := s.workflow.ListRunningDagInstancesByNamePrefix(
		ctx,
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "list running apply dag instances failed")
	}
	if len(applyInstances) > 0 {
		return bizerr.New(bizerr.ErrCodeParams,
			"a create/apply gke heterogeneous deploy operation is in progress for cluster=%d component=%s tidbGroup=%d. "+
				"Wait for it to complete before scaling",
			clusterID, componentType.String(), tidbGroupID)
	}
	return nil
}

func diConcurrencyKey(di *ffentity.DagInstance) string {
	if v, ok := di.Vars[common.KeyGKEHeterogeneousConcurrencyKey]; ok {
		return v.Value
	}
	return ""
}

func gkeHeterogeneousConcurrencyKey(
	clusterID uint64,
	componentType globaldedicatedv1.ComponentType,
	tidbGroupID uint64,
	alternativeInstanceType string,
) string {
	return fmt.Sprintf("%d_%s_%d_%s", clusterID, componentType.String(), tidbGroupID, alternativeInstanceType)
}