package cluster

import (
	"context"
	"encoding/json"
	"strconv"

	"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/entity"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/workflow/action"
	"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/grpc/metadata"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

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

	applyMask := req.GetClusterApplyMask()
	if req.ClusterId == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}
	if isApplyMaskEmpty(applyMask) {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster apply mask not set")
	}

	var cluster *entity.Cluster
	err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		var err error
		cluster, err = s.clusterRepo.Get(ctx, req.ClusterId, false)
		if err != nil {
			return err
		}
		if cluster.State != globaldedicatedv1.Cluster_AVAILABLE {
			return bizerr.New(bizerr.ErrCodeParams, "cluster %d state is %s, cannot apply", cluster.ClusterId, cluster.State.String())
		}

		// Set cluster state to MODIFYING
		cluster.State = globaldedicatedv1.Cluster_MODIFYING
		cluster.HealthState = globaldedicatedv1.Cluster_HEALTHY

		oldVersion := cluster.Version
		if applyMask.GetVersion() {
			// compute target version (even for validate-only)
			cluster.Version, err = s.clusterConfigCli.GetClusterVersion(ctx, cluster)
			if err != nil {
				return err
			}
		}

		if req.ValidateOnly {
			// Do not persist or change state for validation
			return nil
		}

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

		clusterApplyMaskJSON, mErr := protojson.Marshal(applyMask)
		if mErr != nil {
			return bizerr.WrapError(mErr, "marshal cluster apply mask")
		}

		setReq, operations, berr := action.BuildSetVersionAndConfigurationsRequest(
			ctx,
			s.clusterConfigCli,
			cluster,
			applyMask,
		)
		if berr != nil {
			return berr
		}

		setReqJSON, mErr := protojson.Marshal(setReq)
		if mErr != nil {
			return bizerr.WrapError(mErr, "marshal set version & configurations request")
		}

		operationsJSON, mErr := json.Marshal(operations)
		if mErr != nil {
			return bizerr.WrapError(mErr, "marshal operations")
		}

		operationContext := entity.BuildApplyOperationContext(applyMask, operations)
		applyEventBuilder := entity.NewClusterEventBuilder(ctx, cluster, clusterevent.EventTypeApplying)
		applyEventBuilder.SetOperationContext(operationContext)
		if applyMask.GetVersion() {
			applyEventBuilder.SetOldVersion(oldVersion).SetNewVersion(cluster.Version)
		}
		if err := s.eventCommitter.Write(ctx, clusterevent.TopicClusterEvent, applyEventBuilder.Build()); err != nil {
			return err
		}

		metadataJSON, mErr := metadata.MarshalRequestMetaFromCtx(ctx)
		if mErr != nil {
			return bizerr.WrapError(mErr, "marshal metadata")
		}

		operationContextJSON, mErr := json.Marshal(operationContext)
		if mErr != nil {
			return bizerr.WrapError(mErr, "marshal operation context")
		}

		clusterIdStr := strconv.FormatUint(cluster.ClusterId, 10)
		orgIdStr := strconv.FormatUint(cluster.OrganizationId, 10)
		projectIdStr := strconv.FormatUint(cluster.ProjectId, 10)

		if wErr := s.workflow.RunDagWithBizId(
			ctx,
			common.ApplyCluster,
			orgIdStr,
			projectIdStr,
			clusterIdStr,
			map[string]string{
				workflow.KeyTraceId:                          workflow.GetTraceId(ctx),
				workflow.KeyRequestMeta:                      string(metadataJSON),
				workflow.KeyComponentOperations:              string(operationsJSON),
				common.KeyClusterId:                          clusterIdStr,
				common.KeyTargetStatus:                       globaldedicatedv1.Cluster_AVAILABLE.String(),
				common.KeyClusterApplyMask:                   string(clusterApplyMaskJSON),
				common.KeySetVersionAndConfigurationsRequest: string(setReqJSON),
				common.KeyOperationContext:                   string(operationContextJSON),
			},
		); wErr != nil {
			return bizerr.WrapErrorWithCode(wErr, bizerr.ErrCodeUnknown, "create async workflow failed")
		}
		return nil
	})
	if err != nil {
		return nil, bizerr.WrapError(err, "apply cluster failed")
	}

	cluster.HealthState = globaldedicatedv1.Cluster_HEALTHY
	return cluster, nil
}

// Helper: determine if apply mask is effectively empty.
func isApplyMaskEmpty(mask *globaldedicatedv1.ClusterApplyMask) bool {
	if mask == nil {
		return true
	}
	if mask.GetVersion() {
		return false
	}
	return len(mask.GetComponentApplyMasks()) == 0
}