package cron

import (
	"context"
	"fmt"
	"math"
	"strings"
	"sync"

	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/server"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
)

// DefaultRcuScaleDownMaxRuThreshold is the fallback max RU ceiling used
// when the configured value is 0. Above this RU, RCU scale-down leaves a
// component alone: with an unlimited top tier, any RU resolves to the
// highest configured spec, so a component deliberately provisioned above
// that spec would otherwise be clobbered back down.
const DefaultRcuScaleDownMaxRuThreshold uint32 = 50000

// RcuScaleDownJob periodically checks all shared pools and downgrades
// component NodeSpec when total RCU crosses a tier boundary downward.
// The set of components is driven by server.GetRcuScaleComponents().
// Only changes NodeSpec, never NodeCount or disk.
type RcuScaleDownJob struct {
	ngService      server.Service
	lockRepo       repo.LockRepo
	expression     string
	maxRuThreshold uint32
	running        sync.Mutex
}

func NewRcuScaleDownJob(
	ngService server.Service,
	lockRepo repo.LockRepo,
	expression string,
	maxRuThreshold uint32,
) *RcuScaleDownJob {
	return &RcuScaleDownJob{
		ngService:      ngService,
		lockRepo:       lockRepo,
		expression:     expression,
		maxRuThreshold: maxRuThreshold,
	}
}

// effectiveMaxRuThreshold returns the RU ceiling above which scale-down
// skips a component, falling back to the default when unset (e.g. a
// zero-value job in tests or an unset config).
func (j *RcuScaleDownJob) effectiveMaxRuThreshold() uint32 {
	if j.maxRuThreshold == 0 {
		return DefaultRcuScaleDownMaxRuThreshold
	}
	return j.maxRuThreshold
}

func (j *RcuScaleDownJob) Name() string {
	return "RcuScaleDownJob"
}

func (j *RcuScaleDownJob) CronExpression() string {
	return j.expression
}

func (j *RcuScaleDownJob) Execute(ctx context.Context) error {
	if !j.running.TryLock() {
		tilog.Warn(ctx, "rcu_scale_down: job still running, skipping this execution")
		server.AutoscaleRcuRunTotal.WithLabelValues("scale_down", "skip_running").Inc()
		return nil
	}
	defer j.running.Unlock()

	tilog.Info(ctx, "rcu_scale_down: execution started",
		zap.String("cron", j.expression))

	sharedPools, err := j.ngService.ListSharedPools(ctx, nil, nil)
	if err != nil {
		tilog.Error(ctx, "rcu_scale_down: failed to list shared pools", zap.Error(err))
		server.AutoscaleRcuRunTotal.WithLabelValues("scale_down", "error").Inc()
		return fmt.Errorf("rcu_scale_down: failed to list shared pools: %w", err)
	}

	for _, sp := range sharedPools {
		if sp.GetPhase() == apiv1.SharedPool_DELETING {
			continue
		}
		if err := j.processSharedPool(ctx, sp); err != nil {
			tilog.Error(ctx, "rcu_scale_down: failed to process shared pool",
				zap.String("sharedPoolId", sp.GetId()), zap.Error(err))
			server.AutoscaleRcuPoolTotal.WithLabelValues(sp.GetId(), "error").Inc()
		}
	}
	server.AutoscaleRcuRunTotal.WithLabelValues("scale_down", "ok").Inc()
	return nil
}

func (j *RcuScaleDownJob) processSharedPool(ctx context.Context, sp *entity.SharedPool) error {
	if entity.IsEssentialV2(sp) {
		tilog.Info(ctx, "rcu_scale_down: skip Essential v2 shared pool",
			zap.String("sharedPoolId", sp.GetId()))
		server.AutoscaleRcuPoolTotal.WithLabelValues(sp.GetId(), "skip_essential_v2").Inc()
		return nil
	}

	unlockFunc, err := j.lockRepo.Lock(ctx, &entity.Lock{
		Key:         sp.GetId(),
		TimeoutSecs: 10,
	})
	if err != nil {
		tilog.Warn(ctx, "rcu_scale_down: failed to acquire lock, skip",
			zap.String("sharedPoolId", sp.GetId()), zap.Error(err))
		server.AutoscaleRcuPoolTotal.WithLabelValues(sp.GetId(), "skip_lock").Inc()
		return nil
	}
	defer func() { _ = unlockFunc() }()

	ruConfig, err := j.ngService.GetRuConfig(ctx, sp.GetId(), apiv1.RuConfig_SHARED_POOL)
	if err != nil {
		return fmt.Errorf("failed to get ru config: %w", err)
	}

	enableSharedPoolByRu := ruConfig.GetFeatureFlags().GetEnableScaleSharedPoolByRu()
	enableTiDBGroupByRu := ruConfig.GetFeatureFlags().GetEnableScaleTidbGroupByRu()
	tilog.Info(ctx, "rcu_scale_down: loaded feature flags",
		zap.String("sharedPoolId", sp.GetId()),
		zap.Bool("enableScaleSharedPoolByRu", enableSharedPoolByRu),
		zap.Bool("enableScaleTiDBGroupByRu", enableTiDBGroupByRu))
	if !enableSharedPoolByRu && !enableTiDBGroupByRu {
		server.AutoscaleRcuPoolTotal.WithLabelValues(sp.GetId(), "skip_no_flags").Inc()
		return nil
	}

	for _, comp := range sp.GetComponentGroups() {
		if comp.GetPhase() == apiv1.Component_MODIFYING {
			tilog.Info(ctx, "rcu_scale_down: pool has MODIFYING component, skip",
				zap.String("sharedPoolId", sp.GetId()),
				zap.String("componentType", comp.GetComponentType().String()))
			server.AutoscaleRcuPoolTotal.WithLabelValues(sp.GetId(), "skip_modifying").Inc()
			return nil
		}
	}

	tidbClusters, err := j.ngService.ListTiDBClusters(ctx, sp.GetId(), "")
	if err != nil {
		return fmt.Errorf("failed to list tidb clusters: %w", err)
	}

	var totalMaxRu uint64
	for _, tc := range tidbClusters {
		totalMaxRu += uint64(tc.GetRequestUnit().GetMax())
	}
	tilog.Info(ctx, "rcu_scale_down: loaded cluster request units",
		zap.String("sharedPoolId", sp.GetId()),
		zap.Int("clusterCount", len(tidbClusters)),
		zap.Uint64("totalMaxRu", totalMaxRu))
	if totalMaxRu > math.MaxUint32 {
		tilog.Warn(ctx, "rcu_scale_down: total RU overflows uint32, skip",
			zap.String("sharedPoolId", sp.GetId()), zap.Uint64("totalMaxRu", totalMaxRu))
		server.AutoscaleRcuPoolTotal.WithLabelValues(sp.GetId(), "skip_overflow").Inc()
		return nil
	}
	poolMaxRu := uint32(totalMaxRu)

	scaleComponents := make([]*entity.Component, 0)

	for _, cfg := range server.GetRcuScaleComponents() {
		switch cfg.Scope {
		case server.RcuScaleScopeSharedPool:
			if !enableSharedPoolByRu {
				continue
			}
			j.collectSharedPoolDowngrades(ctx, sp, ruConfig, cfg, poolMaxRu, &scaleComponents)

		case server.RcuScaleScopePerCluster:
			if !enableTiDBGroupByRu {
				continue
			}
			j.collectPerClusterDowngrades(ctx, sp, ruConfig, cfg, tidbClusters, &scaleComponents)
		}
	}

	if len(scaleComponents) == 0 {
		tilog.Info(ctx, "rcu_scale_down: no downgrade candidates",
			zap.String("sharedPoolId", sp.GetId()),
			zap.Uint32("poolMaxRu", poolMaxRu))
		server.AutoscaleRcuPoolTotal.WithLabelValues(sp.GetId(), "ok").Inc()
		return nil
	}

	tilog.Info(ctx, "rcu_scale_down: prepared batch scale request",
		zap.String("sharedPoolId", sp.GetId()),
		zap.Uint32("poolMaxRu", poolMaxRu),
		zap.Int("componentCount", len(scaleComponents)),
		zap.String("components", describeScaleComponents(scaleComponents)))

	if _, err := server.ScaleSharedPoolBatchWithLockHeld(ctx, j.ngService, sp.GetId(), scaleComponents, false); err != nil {
		if strings.Contains(err.Error(), "no pods found") {
			tilog.Warn(ctx, "rcu_scale_down: scale protection hit no-pods, retrying with force=true",
				zap.String("sharedPoolId", sp.GetId()))
			if _, retryErr := server.ScaleSharedPoolBatchWithLockHeld(ctx, j.ngService, sp.GetId(), scaleComponents, true); retryErr != nil {
				server.AutoscaleRcuBatchScaleTotal.WithLabelValues(sp.GetId(), "error").Inc()
				return fmt.Errorf("failed to submit batch scale request (force): %w", retryErr)
			}
			tilog.Info(ctx, "rcu_scale_down: batch scale request succeeded after force retry",
				zap.String("sharedPoolId", sp.GetId()),
				zap.Uint32("poolMaxRu", poolMaxRu),
				zap.Int("componentCount", len(scaleComponents)),
				zap.String("components", describeScaleComponents(scaleComponents)),
				zap.Bool("force", true))
			server.AutoscaleRcuBatchScaleTotal.WithLabelValues(sp.GetId(), "success_force").Inc()
			server.AutoscaleRcuPoolTotal.WithLabelValues(sp.GetId(), "ok").Inc()
			return nil
		}
		server.AutoscaleRcuBatchScaleTotal.WithLabelValues(sp.GetId(), "error").Inc()
		return fmt.Errorf("failed to submit batch scale request: %w", err)
	}

	server.AutoscaleRcuBatchScaleTotal.WithLabelValues(sp.GetId(), "success").Inc()
	server.AutoscaleRcuPoolTotal.WithLabelValues(sp.GetId(), "ok").Inc()
	return nil
}

// collectSharedPoolDowngrades checks a single shared-pool-scoped component for downgrade.
func (j *RcuScaleDownJob) collectSharedPoolDowngrades(
	ctx context.Context,
	sp *entity.SharedPool,
	ruConfig *entity.RuConfig,
	cfg server.RcuScaleComponentConfig,
	poolMaxRu uint32,
	scaleComponents *[]*entity.Component,
) {
	if threshold := j.effectiveMaxRuThreshold(); poolMaxRu > threshold {
		tilog.Info(ctx, "rcu_scale_down: skip, pool max RU above threshold",
			zap.String("sharedPoolId", sp.GetId()),
			zap.String("componentType", cfg.ComponentType.String()),
			zap.Uint32("poolMaxRu", poolMaxRu),
			zap.Uint32("threshold", threshold))
		server.AutoscaleRcuScaleDownTotal.WithLabelValues(sp.GetId(), "shared_pool", cfg.ComponentType.String(), "skip_high_ru").Inc()
		return
	}

	targetSpec, err := ruConfig.GetNodeSpec(cfg.ComponentType, poolMaxRu)
	if err != nil {
		tilog.Warn(ctx, "rcu_scale_down: failed to get target spec",
			zap.String("sharedPoolId", sp.GetId()),
			zap.String("componentType", cfg.ComponentType.String()),
			zap.Error(err))
		return
	}

	comp := findDefaultComponent(sp, cfg.ComponentType)
	if comp == nil {
		return
	}

	currentSpec := comp.GetNodeQuota().GetNodeSpec()
	if !needsDowngrade(currentSpec, targetSpec) {
		return
	}
	if cfg.ScaleOverride != nil && !cfg.ScaleOverride(currentSpec, targetSpec) {
		tilog.Info(ctx, "rcu_scale_down: override vetoed downgrade",
			zap.String("sharedPoolId", sp.GetId()),
			zap.String("componentType", cfg.ComponentType.String()),
			zap.String("componentId", comp.GetId()),
			zap.Uint32("currentCpu", currentSpec.GetCpu()),
			zap.Uint32("targetCpu", targetSpec.GetCpu()))
		server.AutoscaleRcuScaleDownTotal.WithLabelValues(sp.GetId(), "shared_pool", cfg.ComponentType.String(), "vetoed").Inc()
		return
	}

	tilog.Info(ctx, "rcu_scale_down: downgrading pool component",
		zap.String("sharedPoolId", sp.GetId()),
		zap.String("componentType", cfg.ComponentType.String()),
		zap.String("componentId", comp.GetId()),
		zap.Uint32("currentCpu", currentSpec.GetCpu()),
		zap.Uint32("currentMem", currentSpec.GetMemoryGi()),
		zap.Uint32("targetCpu", targetSpec.GetCpu()),
		zap.Uint32("targetMem", targetSpec.GetMemoryGi()),
		zap.Uint32("poolMaxRu", poolMaxRu))
	server.AutoscaleRcuScaleDownTotal.WithLabelValues(sp.GetId(), "shared_pool", cfg.ComponentType.String(), "downgrading").Inc()

	*scaleComponents = append(*scaleComponents, cloneScaleComponent(comp, targetSpec, cfg.ComponentType))
}

// collectPerClusterDowngrades checks a per-cluster-scoped component for downgrade
// across all TiDB clusters in the shared pool.
func (j *RcuScaleDownJob) collectPerClusterDowngrades(
	ctx context.Context,
	sp *entity.SharedPool,
	ruConfig *entity.RuConfig,
	cfg server.RcuScaleComponentConfig,
	tidbClusters []*entity.TiDBCluster,
	scaleComponents *[]*entity.Component,
) {
	for _, tc := range tidbClusters {
		clusterMaxRu := tc.GetRequestUnit().GetMax()

		targetSpec, err := ruConfig.GetNodeSpec(cfg.ComponentType, clusterMaxRu)
		if err != nil {
			tilog.Warn(ctx, "rcu_scale_down: failed to get target spec",
				zap.String("sharedPoolId", sp.GetId()),
				zap.String("clusterId", tc.GetId()),
				zap.String("componentType", cfg.ComponentType.String()),
				zap.Error(err))
			continue
		}

		// For TIDB, iterate each TiDB group within the cluster.
		if cfg.ComponentType == apiv1.ComponentType_TIDB {
			tidbGroups, err := j.ngService.ListTiDBGroups(ctx, tc.GetId())
			if err != nil {
				tilog.Warn(ctx, "rcu_scale_down: failed to list tidb groups",
					zap.String("sharedPoolId", sp.GetId()),
					zap.String("clusterId", tc.GetId()),
					zap.Error(err))
				continue
			}
			for _, tg := range tidbGroups {
				comp := findComponentByTypeAndId(sp, cfg.ComponentType, tg.GetId())
				if comp == nil {
					continue
				}
				j.checkAndAppendDowngrade(ctx, sp.GetId(), tc.GetId(), cfg, comp, targetSpec, clusterMaxRu, scaleComponents)
			}
		} else {
			// Non-TIDB per-cluster components are looked up by cluster ID.
			// Currently TIDB is the only PerCluster component, so this branch
			// is unreachable — kept for forward compatibility.
			comp := findComponentByTypeAndId(sp, cfg.ComponentType, tc.GetId())
			if comp == nil {
				continue
			}
			j.checkAndAppendDowngrade(ctx, sp.GetId(), tc.GetId(), cfg, comp, targetSpec, clusterMaxRu, scaleComponents)
		}
	}
}

func (j *RcuScaleDownJob) checkAndAppendDowngrade(
	ctx context.Context,
	sharedPoolId string,
	clusterId string,
	cfg server.RcuScaleComponentConfig,
	comp *entity.Component,
	targetSpec *entity.NodeSpec,
	ru uint32,
	scaleComponents *[]*entity.Component,
) {
	if threshold := j.effectiveMaxRuThreshold(); ru > threshold {
		tilog.Info(ctx, "rcu_scale_down: skip, cluster max RU above threshold",
			zap.String("sharedPoolId", sharedPoolId),
			zap.String("clusterId", clusterId),
			zap.String("componentType", cfg.ComponentType.String()),
			zap.String("componentId", comp.GetId()),
			zap.Uint32("ru", ru),
			zap.Uint32("threshold", threshold))
		server.AutoscaleRcuScaleDownTotal.WithLabelValues(sharedPoolId, "per_cluster", cfg.ComponentType.String(), "skip_high_ru").Inc()
		return
	}

	currentSpec := comp.GetNodeQuota().GetNodeSpec()
	if !needsDowngrade(currentSpec, targetSpec) {
		return
	}
	if cfg.ScaleOverride != nil && !cfg.ScaleOverride(currentSpec, targetSpec) {
		fields := []zap.Field{
			zap.String("sharedPoolId", sharedPoolId),
			zap.String("clusterId", clusterId),
			zap.String("componentType", cfg.ComponentType.String()),
			zap.String("componentId", comp.GetId()),
			zap.Uint32("currentCpu", currentSpec.GetCpu()),
			zap.Uint32("targetCpu", targetSpec.GetCpu()),
		}
		if cfg.ComponentType == apiv1.ComponentType_TIDB {
			fields = append(fields, zap.String("tidbGroupId", comp.GetId()))
		}
		tilog.Info(ctx, "rcu_scale_down: override vetoed downgrade", fields...)
		server.AutoscaleRcuScaleDownTotal.WithLabelValues(sharedPoolId, "per_cluster", cfg.ComponentType.String(), "vetoed").Inc()
		return
	}

	fields := []zap.Field{
		zap.String("sharedPoolId", sharedPoolId),
		zap.String("clusterId", clusterId),
		zap.String("componentType", cfg.ComponentType.String()),
		zap.String("componentId", comp.GetId()),
		zap.Uint32("currentCpu", currentSpec.GetCpu()),
		zap.Uint32("currentMem", currentSpec.GetMemoryGi()),
		zap.Uint32("targetCpu", targetSpec.GetCpu()),
		zap.Uint32("targetMem", targetSpec.GetMemoryGi()),
		zap.Uint32("ru", ru),
	}
	if cfg.ComponentType == apiv1.ComponentType_TIDB {
		fields = append(fields, zap.String("tidbGroupId", comp.GetId()))
	}
	tilog.Info(ctx, "rcu_scale_down: downgrading component", fields...)
	server.AutoscaleRcuScaleDownTotal.WithLabelValues(sharedPoolId, "per_cluster", cfg.ComponentType.String(), "downgrading").Inc()

	*scaleComponents = append(*scaleComponents, cloneScaleComponent(comp, targetSpec, cfg.ComponentType))
}

func cloneScaleComponent(comp *entity.Component, targetSpec *entity.NodeSpec, compType apiv1.ComponentType) *entity.Component {
	return &entity.Component{
		Id:            comp.GetId(),
		ComponentType: compType,
		NodeQuota: &entity.NodeQuota{
			NodeSpec:     targetSpec,
			NodeCount:    comp.GetNodeQuota().GetNodeCount(),
			MaxNodeCount: comp.GetNodeQuota().GetMaxNodeCount(),
			MinNodeCount: comp.GetNodeQuota().GetMinNodeCount(),
		},
	}
}

// findDefaultComponent finds the default component (id = fnvHash of sharedPoolId) for a given type.
func findDefaultComponent(sp *entity.SharedPool, compType apiv1.ComponentType) *entity.Component {
	defaultId := utils.FnvHash(sp.GetId())
	return findComponentByTypeAndId(sp, compType, defaultId)
}

// findComponentByTypeAndId finds a component matching both type and id.
func findComponentByTypeAndId(sp *entity.SharedPool, compType apiv1.ComponentType, id string) *entity.Component {
	for _, comp := range sp.GetComponentGroups() {
		if comp.GetComponentType() == compType && comp.GetId() == id {
			return comp
		}
	}
	return nil
}

// needsDowngrade returns true if current spec is strictly larger than target spec
// in both CPU and memory dimensions. A mixed change (one dimension up, one down)
// is not considered a downgrade.
//
// NOTE: This assumes that RU tier configs always change CPU and memory together.
// Components without a ScaleOverride (e.g. TIDB_WORKER, COPROCESSOR_WORKER) rely
// on this assumption. If a future tier config changes only one dimension, add a
// ScaleOverride for that component (similar to TiKV's tikvCPUChangeOnly).
func needsDowngrade(current, target *entity.NodeSpec) bool {
	if current == nil || target == nil {
		return false
	}
	return current.GetCpu() > target.GetCpu() && current.GetMemoryGi() > target.GetMemoryGi()
}

func describeScaleComponents(scaleComponents []*entity.Component) string {
	if len(scaleComponents) == 0 {
		return ""
	}
	result := ""
	for idx, comp := range scaleComponents {
		if idx > 0 {
			result += "; "
		}
		result += fmt.Sprintf("%s/%s cpu=%d memGi=%d nodeCount=%d",
			comp.GetComponentType().String(),
			comp.GetId(),
			comp.GetNodeQuota().GetNodeSpec().GetCpu(),
			comp.GetNodeQuota().GetNodeSpec().GetMemoryGi(),
			comp.GetNodeQuota().GetNodeCount(),
		)
	}
	return result
}