package cluster

import (
	"context"
	"strconv"
	"strings"
	"time"

	"github.com/samber/lo"
	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/timestamppb"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	clusterrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/cluster"
	"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/rcuquota"
	"github.com/tidbcloud/tidb-management-service/nextgen-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/workflow"
)

func (s *serviceImpl) Update(
	ctx context.Context,
	req *globalngv1.UpdateClusterRequest,
) (*entity.Cluster, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "ClusterService.Update")
	defer span.End()

	if req.Cluster == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "empty request body")
	}

	var (
		cluster   *entity.Cluster
		err       error
		oldMaxRcu int64
		newMaxRcu int64
	)

	if terr := s.transaction.Transaction(ctx, func(txCtx context.Context) error {
		cluster, err = s.clusterRepo.Get(ctx, req.Cluster.ClusterId, false)
		if err != nil {
			return bizerr.WrapGormError(err, "load cluster failed")
		}
		if cluster == nil {
			return bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster not found")
		}
		resourcePoolID, parseErr := strconv.ParseUint(cluster.GetSharedPoolId(), 10, 64)
		if parseErr != nil || resourcePoolID == 0 {
			return bizerr.New(bizerr.ErrCodeUnknown, "cluster has an invalid resource pool id")
		}
		resourcePool, loadErr := s.resourcePoolRepo.Load(txCtx, resourcePoolID, false)
		if loadErr != nil {
			return loadErr
		}
		if lifecycleErr := servicecommon.RequireResourcePoolLifecycle(
			resourcePool,
			resourcepoolpolicy.OperationUpdateInstance,
		); lifecycleErr != nil {
			return lifecycleErr
		}
		if haErr := validateImmutableHighAvailabilityTypeForUpdate(req.GetCluster(), cluster); haErr != nil {
			return haErr
		}
		oldMaxRcu = cluster.GetMaxRcu()
		newMaxRcu = req.GetCluster().GetMaxRcu()

		ruVersionChanged := req.GetCluster().GetRuVersion() > 0 && req.GetCluster().GetRuVersion() != cluster.RuVersion
		// Only consider RCU changed when MaxRcu is explicitly provided (> 0).
		// When only ru_version is being updated (e.g., via updateRuVersion API),
		// MaxRcu is 0 (proto zero value) and should not trigger RCU changes.
		rcuChanged := req.GetCluster().GetMaxRcu() > 0 &&
			(req.GetCluster().GetMaxRcu() != cluster.GetMaxRcu() || req.GetCluster().GetMinRcu() != cluster.GetMinRcu())
		reservedCapacityTemplateChanged := req.GetCluster().GetReservedCapacityTemplateId() > 0 &&
			req.GetCluster().GetReservedCapacityTemplateId() != cluster.GetReservedCapacityTemplateId()

		if !rcuChanged && !ruVersionChanged && !reservedCapacityTemplateChanged {
			tilog.Info(ctx, "no need to update cluster, skipping", zap.Uint64("cluster_id", cluster.ClusterId))
			cluster.State = globalngv1.Cluster_AVAILABLE
			cluster.HealthState = globalngv1.Cluster_HEALTHY
			return nil
		}

		if cluster.State != globalngv1.Cluster_AVAILABLE {
			return bizerr.New(bizerr.ErrCodeParams, "cluster is not available")
		}

		if rcuChanged {
			rcuQuota, err := s.rcuQuotaRepo.Load(ctx, req.GetCluster().GetOrganizationId(), req.GetCluster().ServicePlan.String())
			if err != nil {
				return bizerr.WrapError(err, "load rcu quota failed")
			}

			req.GetCluster().MinRcu = rcuquota.CalculateMinRcuByQuota(req.GetCluster().GetMaxRcu(), rcuQuota)

			if req.GetCluster().GetMaxRcu() > int64(rcuQuota.InstanceMaxRcu) {
				return bizerr.New(
					bizerr.ErrCodeParams,
					"The total RCU limit cannot exceed %d for the organization.",
					rcuQuota.InstanceMaxRcu,
				)
			}

			if req.GetCluster().GetMinRcu() > req.GetCluster().GetMaxRcu() {
				return bizerr.New(bizerr.ErrCodeParams, "min RCU must be less than or equal to max RCU")
			}

			// validate total rcu of org
			clusters, err := s.clusterRepo.Query(ctx, &clusterrepo.Filter{
				OrganizationIds: []uint64{cluster.GetOrganizationId()},
				ServicePlans:    []entity.ServicePlan{cluster.GetServicePlan()},
			})

			if err != nil {
				return bizerr.WrapError(err, "query clusters for rcu validation failed")
			}

			totalMaxRcu := lo.Reduce(clusters, func(agg uint64, item *entity.Cluster, index int) uint64 {
				return agg + uint64(item.GetMaxRcu())
			}, 0)

			// Increment
			if uint64(req.Cluster.MaxRcu-cluster.MaxRcu)+totalMaxRcu > rcuQuota.TotalMaxRcu {
				return bizerr.New(
					bizerr.ErrCodeParams,
					"The total RCU limit cannot exceed %d for the organization.",
					rcuQuota.TotalMaxRcu,
				)
			}

			region, err := s.regionRepo.Get(ctx, cluster.RegionId)
			if err != nil {
				return bizerr.WrapGormError(err, "load region failed")
			}
			if region == nil {
				return bizerr.New(bizerr.ErrCodeRecordNotFound, "region not found")
			}
			tidbgroups, err := s.tidbgroupRepo.Query(ctx, &tidbgroup.Filter{
				ClusterIds: []uint64{cluster.ClusterId},
			})
			if err != nil {
				return bizerr.WrapGormError(err, "query tidbgroup failed")
			}
			for _, tg := range tidbgroups {
				regionalTidbgroup, err := s.regionalCli.GetTidbGroup(ctx, region, &clusterngv1beta.GetTiDBGroupRequest{
					ClusterId:   strconv.FormatUint(cluster.ClusterId, 10),
					TidbGroupId: strconv.FormatUint(tg.Id, 10),
				})
				if err != nil {
					return bizerr.WrapError(err, "get regional tidbgroup failed")
				}
				if regionalTidbgroup.Phase == clusterngv1beta.TiDBGroup_UNAVAILABLE {
					return bizerr.New(bizerr.ErrCodeParams, "tidbgroup is not running")
				}
			}

			// V2 Cooldown logic for Max RCU Scale Down
			if req.GetCluster().GetMaxRcu() < cluster.GetMaxRcu() {
				if cluster.CapacityUpdatedAt != nil {
					cooldownMinutes := rcuQuota.ScaleDownCooldownMinutes
					cooldown := time.Duration(cooldownMinutes) * time.Minute
					timeSinceLastUpdate := time.Since(*cluster.CapacityUpdatedAt)
					if timeSinceLastUpdate < cooldown {
						return bizerr.New(bizerr.ErrCodeParams,
							"Scale-down is currently unavailable. You will be able to decrease the capacity limit after %s",
							cluster.CapacityUpdatedAt.Add(cooldown).UTC().Format("Jan 2, 3:04 PM MST"))
					}
				}
			}

			if req.GetCluster().GetMinRcu() != cluster.GetMinRcu() {
				cluster.MinRcu = req.GetCluster().GetMinRcu()
			}
			if req.GetCluster().GetMaxRcu() != cluster.GetMaxRcu() {
				cluster.MaxRcu = req.GetCluster().GetMaxRcu()
				now := time.Now()
				cluster.CapacityUpdatedAt = &now
				cluster.Cluster.CapacityUpdatedAt = timestamppb.New(now)
			}
		}
		// update ru_version if set (> 0)
		if req.GetCluster().GetRuVersion() > 0 {
			cluster.RuVersion = req.GetCluster().GetRuVersion()
		}
		previousReservedCapacityTemplateId := cluster.GetReservedCapacityTemplateId()
		if reservedCapacityTemplateChanged {
			if cluster.GetServicePlan() != globalcommonv1.ServicePlan_Premium_Reserved {
				return bizerr.New(bizerr.ErrCodeParams, "reserved_capacity_template_id is only supported for premium reserved clusters")
			}

			template, err := s.reservedCapacityTemplateRepo.Load(txCtx, req.GetCluster().GetReservedCapacityTemplateId(), false)
			if err != nil {
				return bizerr.WrapError(err, "load reserved capacity template failed")
			}
			if template.GetStatus() != globalngv1.ReservedCapacityTemplate_ACTIVE {
				return bizerr.New(bizerr.ErrCodeParams, "reserved capacity template is not active")
			}
			if template.GetScope() == globalngv1.ReservedCapacityTemplate_ORG_PRIVATE &&
				template.GetOrganizationId() != cluster.GetOrganizationId() {
				return bizerr.New(bizerr.ErrCodeParams, "reserved capacity template does not belong to organization")
			}
			if template.GetScope() != globalngv1.ReservedCapacityTemplate_DEFAULT &&
				template.GetScope() != globalngv1.ReservedCapacityTemplate_ORG_PRIVATE {
				return bizerr.New(bizerr.ErrCodeParams, "reserved capacity template scope is invalid")
			}

			cluster.ReservedCapacityTemplateId = req.GetCluster().GetReservedCapacityTemplateId()
			now := time.Now()
			cluster.CapacityUpdatedAt = &now
			cluster.Cluster.CapacityUpdatedAt = timestamppb.New(now)
		}
		if req.GetValidateOnly() {
			return nil
		}
		lockedResourcePool, lockErr := s.resourcePoolRepo.LoadForUpdate(txCtx, resourcePoolID, false)
		if lockErr != nil {
			return lockErr
		}
		if lifecycleErr := servicecommon.RequireResourcePoolLifecycle(
			lockedResourcePool,
			resourcepoolpolicy.OperationUpdateInstance,
		); lifecycleErr != nil {
			return lifecycleErr
		}
		cluster.State = globalngv1.Cluster_MODIFYING
		if err := s.clusterRepo.Save(txCtx, cluster); err != nil {
			return bizerr.WrapGormError(err, "save cluster failed")
		}

		if rcuChanged {
			ev, err := entity.NewClusterEvent(ctx, cluster, clusterevent.EventTypeUpdatingCapacity, "CLUSTER_STATUS_MODIFYING")
			if err != nil {
				return err
			}
			if err := s.eventCommitter.Write(ctx, clusterevent.TopicClusterEvent, ev); err != nil {
				return err
			}
		}

		dagID := common.UpdateCluster
		if reservedCapacityTemplateChanged {
			dagID = common.UpdateClusterTopology
		}

		tidbGroupIds := ""
		if reservedCapacityTemplateChanged {
			if s.tidbgroupRepo != nil {
				tidbgroups, qErr := s.tidbgroupRepo.Query(ctx, &tidbgroup.Filter{ClusterIds: []uint64{cluster.ClusterId}})
				if qErr != nil {
					return bizerr.WrapGormError(qErr, "query tidbgroup failed")
				}
				ids := make([]string, 0, len(tidbgroups))
				for _, tg := range tidbgroups {
					ids = append(ids, strconv.FormatUint(tg.Id, 10))
				}
				tidbGroupIds = strings.Join(ids, ",")
			}
		}

		if err := s.workflow.RunDagWithBizId(
			ctx,
			dagID,
			strconv.FormatUint(cluster.OrganizationId, 10),
			"0",
			strconv.FormatUint(cluster.ClusterId, 10),
			map[string]string{
				workflow.KeyTraceId:      workflow.GetTraceId(ctx),
				common.KeyClusterId:      strconv.FormatUint(cluster.ClusterId, 10),
				common.KeyTidbGroupIds:   tidbGroupIds,
				common.KeyResourcePoolId: cluster.GetSharedPoolId(),
				common.KeyTargetStatus:   globalngv1.Cluster_AVAILABLE.String(),
				common.KeyOldMaxRcu:      strconv.FormatInt(oldMaxRcu, 10),
				common.KeyNewMaxRcu:      strconv.FormatInt(newMaxRcu, 10),
				common.KeyEventName: string(
					getUpdateClusterDoneEventName(rcuChanged, ruVersionChanged, reservedCapacityTemplateChanged),
				),
				common.KeyPreviousReservedCapacityTemplateId: strconv.FormatUint(previousReservedCapacityTemplateId, 10),
				common.KeyReservedCapacityTemplateId:         strconv.FormatUint(req.GetCluster().GetReservedCapacityTemplateId(), 10),
			},
		); err != nil {
			return bizerr.WrapError(err, "run workflow failed")
		}
		return nil
	}); terr != nil {
		return nil, bizerr.WrapError(terr, "update cluster transaction failed")
	}

	return cluster, nil
}

func validateImmutableHighAvailabilityTypeForUpdate(
	requested *globalngv1.Cluster,
	current *entity.Cluster,
) bizerr.BizError {
	requestedType := requested.GetHighAvailabilityType()
	if requestedType == globalngv1.Cluster_HIGH_AVAILABILITY_TYPE_UNSPECIFIED {
		return nil
	}

	currentType := current.GetHighAvailabilityType()
	if currentType == globalngv1.Cluster_HIGH_AVAILABILITY_TYPE_UNSPECIFIED {
		currentType = globalngv1.Cluster_REGIONAL
	}
	if requestedType != currentType {
		return bizerr.New(bizerr.ErrCodeParams, "high_availability_type is immutable")
	}

	return nil
}

func getUpdateClusterDoneEventName(rcuChanged, ruVersionChanged, reservedCapacityTemplateChanged bool) clusterevent.EventType {
	if rcuChanged {
		return clusterevent.EventTypeUpdatedCapacity
	}
	if ruVersionChanged || reservedCapacityTemplateChanged {
		return clusterevent.EventTypeApplied
	}
	return clusterevent.EventTypeNormal
}