package server

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	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/pkg/autoscale/spec"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"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"
	"google.golang.org/protobuf/types/known/structpb"
	"google.golang.org/protobuf/types/known/wrapperspb"
)

// defaultAutoscalePolicySubjectID is the singleton subject_id for the
// SCOPE_DEFAULT row in the policy bundle table.
const defaultAutoscalePolicySubjectID = "__default__"

// unwrapBoolValue extracts the *bool from a *wrapperspb.BoolValue (nil-safe).
func unwrapBoolValue(b *wrapperspb.BoolValue) *bool {
	if b == nil {
		return nil
	}
	v := b.GetValue()
	return &v
}

// wrapBoolValue converts a *bool to a *wrapperspb.BoolValue (nil-safe).
func wrapBoolValue(b *bool) *wrapperspb.BoolValue {
	if b == nil {
		return nil
	}
	return wrapperspb.Bool(*b)
}

// validateImplementedPolicyScope is the per-scope gate on every RPC
// entry point. Accepts DEFAULT / TENANT / SHARED_POOL / CLUSTER;
// rejects UNSPECIFIED and any non-implemented enum value. CLUSTER
// has no bundle row — see design doc §Inheritance Chains.
func validateImplementedPolicyScope(scope apiv1.AutoscalePolicyScope) error {
	switch scope {
	case apiv1.AutoscalePolicyScope_SCOPE_DEFAULT,
		apiv1.AutoscalePolicyScope_SCOPE_TENANT,
		apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL,
		apiv1.AutoscalePolicyScope_SCOPE_TIDB_GROUP:
		return nil
	default:
		return status.Errorf(codes.InvalidArgument, "autoscale policy scope is unspecified or invalid: %s", scope.String())
	}
}

// resolveSubjectID picks the subject_id for a scope. DEFAULT always
// uses defaultAutoscalePolicySubjectID; other scopes require a
// non-empty caller-supplied id.
func resolveSubjectID(scope apiv1.AutoscalePolicyScope, subjectID string) (string, error) {
	if scope == apiv1.AutoscalePolicyScope_SCOPE_DEFAULT {
		return defaultAutoscalePolicySubjectID, nil
	}
	if subjectID == "" {
		return "", status.Errorf(codes.InvalidArgument, "subject_id is required for scope %s", scope.String())
	}
	return subjectID, nil
}

// hashBundle returns a stable hash of the canonical bundle (used for
// SYNC KEYS_ONLY change detection).
func hashBundle(specs []*entity.AutoscalePolicySpec) string {
	// Filter nil entries so sort/marshal never deref a nil pointer.
	filtered := make([]*entity.AutoscalePolicySpec, 0, len(specs))
	for _, s := range specs {
		if s != nil {
			filtered = append(filtered, s)
		}
	}
	sorted := make([]*entity.AutoscalePolicySpec, len(filtered))
	copy(sorted, filtered)
	sort.Slice(sorted, func(i, j int) bool {
		a, b := sorted[i], sorted[j]
		if a.GetTargetType() != b.GetTargetType() {
			return a.GetTargetType() < b.GetTargetType()
		}
		if a.GetDirection() != b.GetDirection() {
			return a.GetDirection() < b.GetDirection()
		}
		if a.GetCapability() != b.GetCapability() {
			return a.GetCapability() < b.GetCapability()
		}
		return a.GetTrigger() < b.GetTrigger()
	})
	h := sha256.New()
	for _, s := range sorted {
		// protojson → stable text serialization independent of map order.
		b, _ := protojson.Marshal(s)
		h.Write(b)
		h.Write([]byte("\n"))
	}
	return hex.EncodeToString(h.Sum(nil))[:32]
}

// ---------------------------------------------------------------------------
// Write RPCs
// ---------------------------------------------------------------------------

// CreateOrReplaceAutoscalePolicy implements POST (row-level REPLACE).
// Caller's specs become the entire canonical bundle. No DELETE / CLEAR
// RPC — to clear, POST an empty spec list. See design doc §Field Semantics.
func (s *serverImpl) CreateOrReplaceAutoscalePolicy(
	ctx context.Context,
	scope apiv1.AutoscalePolicyScope,
	subjectID string,
	specs []*entity.AutoscalePolicySpec,
	applyToAllSharedPools *wrapperspb.BoolValue,
	requestID string,
) (resp *entity.AutoscalePolicyResponse, err error) {
	// Metric captures both success and early-reject paths so dashboards
	// distinguish "operator error" from "Internal failure".
	// See design doc §Metrics invariants.
	startedAt := time.Now()
	defer func() {
		markPolicyWrite(scope, operationCreateOrReplace, applyToAllSharedPools,
			statusCodeToPolicyWriteResult(status.Code(err)), startedAt)
	}()

	if err := validateImplementedPolicyScope(scope); err != nil {
		return nil, err
	}
	resolvedID, err := resolveSubjectID(scope, subjectID)
	if err != nil {
		return nil, err
	}

	// apply_to_all_shared_pools=true is only valid at TENANT scope (see
	// proto docs services.proto:2420-2422).
	if err := validateApplyToAllSharedPoolsScope(scope, operationCreateOrReplace, applyToAllSharedPools); err != nil {
		return nil, err
	}
	if err := validateBundleScopeComponentCompatibility(scope, specs); err != nil {
		return nil, err
	}

	// TIDB_GROUP scope governs per-group TIDB and TIPROXY HPA behavior.
	// Workers by the compatibility check
	// above; workers → SHARED_POOL and TIKV scale-in/out →
	// ru_configs.FeatureFlags at SHARED_POOL scope.
	if scope == apiv1.AutoscalePolicyScope_SCOPE_TIDB_GROUP {
		if bad := nonClusterComponentsInSpecs(specs); len(bad) > 0 {
			return nil, status.Errorf(codes.InvalidArgument,
				"TIDB_GROUP scope only accepts TIDB components; unsupported: %s", strings.Join(bad, ","))
		}
		if bad := nonClusterSpecKeysInSpecs(specs); len(bad) > 0 {
			return nil, status.Errorf(codes.InvalidArgument,
				"TIDB_GROUP scope only accepts TIDB at SCALE_IN (NODE_COUNT, CPU) or SCALE_OUT (NODE_COUNT, CPU); unsupported keys: %s", strings.Join(bad, ","))
		}
		// POST = REPLACE: cannot actually clear HPA state at CLUSTER
		// scope (see design doc §TIDB_GROUP scope) — reject empty POSTs
		// rather than silently returning SKIPPED with old HPA intact.
		if len(specs) == 0 {
			return nil, status.Errorf(codes.InvalidArgument,
				"TIDB_GROUP scope POST requires non-empty specs; empty POST would silently preserve HPA state (clear/replace not implemented)")
		}
	}

	// SHARED_POOL scope rejects components/keys outside the mapping
	// table — see design doc §Core Configuration Mapping.
	if scope == apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL {
		if bad := nonSharedPoolComponentsInSpecs(specs); len(bad) > 0 {
			return nil, status.Errorf(codes.InvalidArgument,
				"SHARED_POOL scope only accepts TIKV / TIKV_WORKER / TIDB_WORKER / COPROCESSOR_WORKER / SHARED_POOL / TIDB_GROUP components; unsupported: %s", strings.Join(bad, ","))
		}
		if bad := nonSharedPoolSpecKeysInSpecs(specs); len(bad) > 0 {
			return nil, status.Errorf(codes.InvalidArgument,
				"SHARED_POOL scope spec key outside mapping table; unsupported keys: %s", strings.Join(bad, ","))
		}
	}

	// SHARED_POOL POST with no specs is a no-op, not a dense replace with
	// an all-nil FeatureFlags block. Use SetRuConfig directly when an
	// operator intentionally needs to wipe FeatureFlags overrides.
	if scope == apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL && len(specs) == 0 {
		return &entity.AutoscalePolicyResponse{
			Targets: []*entity.AutoscalePolicyApplyTarget{{
				Scope:     scope,
				SubjectId: resolvedID,
				Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SKIPPED,
				Message:   "SHARED_POOL POST with empty specs is a no-op (not a FeatureFlag wipe); if you intend to wipe, use SetRuConfig RPC directly",
			}},
		}, nil
	}

	// POST = REPLACE: rejects nil/UNSPECIFIED state — see design doc
	// §Field Semantics ("Passing nil values"). PATCH is the path for
	// "inherit from parent scope" semantics.
	for i, s := range specs {
		if s == nil {
			return nil, status.Errorf(codes.InvalidArgument,
				"POST does not accept nil specs (specs[%d]); PATCH is the path for unset/inherit semantics", i)
		}
		if s.GetState() == apiv1.AutoscalePolicy_STATE_UNSPECIFIED {
			return nil, status.Errorf(codes.InvalidArgument,
				"POST does not accept nil state (specs[%d] = %s/%s/%s); use PATCH to inherit, or set an explicit ENABLED/DISABLED",
				i, s.GetTargetType(), s.GetDirection(), s.GetCapability())
		}
	}

	// DEFAULT scope: empty POST is a no-op (not a wipe) — see design doc
	// §DEFAULT Scope Specifics.
	if scope == apiv1.AutoscalePolicyScope_SCOPE_DEFAULT && len(specs) == 0 {
		return &entity.AutoscalePolicyResponse{
			Targets: []*entity.AutoscalePolicyApplyTarget{
				{
					SubjectId: resolvedID,
					Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SKIPPED,
					Message:   "DEFAULT scope POST specs is empty; no changes to apply (POST specs:[] is a no-op)",
				},
			},
		}, nil
	}

	// POST = REPLACE: build the canonical bundle as the user-supplied
	// specs, then upsert.
	merged := spec.MergeSpecs(nil, specs, spec.MergeModeReplace)
	hash := hashBundle(merged)

	row := &repo.AutoscalePolicyBundleRow{
		Scope:                 scope,
		SubjectID:             resolvedID,
		Specs:                 merged,
		ApplyToAllSharedPools: unwrapBoolValue(applyToAllSharedPools),
		BundleHash:            hash,
	}
	// SHARED_POOL and CLUSTER store policy in HPA + ru_config.featureFlags,
	// not the bundle table. DEFAULT/TENANT use the bundle table.
	// See design doc §Inheritance Chains.
	if scope == apiv1.AutoscalePolicyScope_SCOPE_DEFAULT ||
		scope == apiv1.AutoscalePolicyScope_SCOPE_TENANT {
		if err := s.autoscalePolicyBundleRepo.UpsertBundle(ctx, row); err != nil {
			return nil, status.Errorf(codes.Internal, "upsert policy bundle: %v", err)
		}
	}

	// Translate to ru_configs.FeatureFlags patch + apply per-target.
	// POST semantics: the request IS the new full state, so
	// `merged` (the post-REPLACE canonical bundle) is also the
	// delta. The materialize helper uses FULL-REPLACE mode for
	// POST (see materializeApplyMode), so passing the full
	// bundle as delta is correct: every key in the bundle
	// replaces the corresponding existing row, and any key NOT
	// in the request is dropped (the canonical bundle is
	// authoritative for POST).
	delta := spec.DeltaSpecs(merged)
	applyCtx := policyApplyContext{
		scope:                 scope,
		subjectID:             resolvedID,
		applyToAllSharedPools: applyToAllSharedPools,
		requestID:             requestID,
		operation:             operationCreateOrReplace,
	}
	targets, err := s.applyPolicyPatchToTargets(ctx, applyCtx, delta, merged)
	if err != nil {
		return nil, err
	}
	return &entity.AutoscalePolicyResponse{Targets: targets}, nil
}

// PatchAutoscalePolicy implements PATCH (row-level MERGE: per-key
// upsert with last-writer-wins). State=UNSPECIFIED carries no value.
//
// TODO(autoscale-policy): PATCH cannot selectively remove a single
// override key today — passing State=UNSPECIFIED is a no-op (the key
// is preserved in the merged bundle, not deleted). Operators must
// POST the full desired bundle to drop a key. A future
// DeleteAutoscalePolicy / ClearAutoscalePolicyBundle RPC (design doc
// §Open Question #14) will support per-key removal. Tracked in
// pkg/autoscale/spec/spec.go MergeModeMerge contract as well.
func (s *serverImpl) PatchAutoscalePolicy(
	ctx context.Context,
	scope apiv1.AutoscalePolicyScope,
	subjectID string,
	specs []*entity.AutoscalePolicySpec,
	applyToAllSharedPools *wrapperspb.BoolValue,
	requestID string,
) (resp *entity.AutoscalePolicyResponse, err error) {
	// write_total + write_duration metric — same pattern as Create.
	startedAt := time.Now()
	defer func() {
		markPolicyWrite(scope, operationPatch, applyToAllSharedPools,
			statusCodeToPolicyWriteResult(status.Code(err)), startedAt)
	}()

	if err := validateImplementedPolicyScope(scope); err != nil {
		return nil, err
	}
	resolvedID, err := resolveSubjectID(scope, subjectID)
	if err != nil {
		return nil, err
	}

	if err := validateApplyToAllSharedPoolsScope(scope, operationPatch, applyToAllSharedPools); err != nil {
		return nil, err
	}
	if err := validateBundleScopeComponentCompatibility(scope, specs); err != nil {
		return nil, err
	}

	// TIDB_GROUP scope governs per-group TIDB and TIPROXY HPA behavior. See design doc §Core Configuration Mapping.
	if scope == apiv1.AutoscalePolicyScope_SCOPE_TIDB_GROUP {
		if bad := nonClusterComponentsInSpecs(specs); len(bad) > 0 {
			return nil, status.Errorf(codes.InvalidArgument,
				"TIDB_GROUP scope only accepts TIDB components; unsupported: %s", strings.Join(bad, ","))
		}
		if bad := nonClusterSpecKeysInSpecs(specs); len(bad) > 0 {
			return nil, status.Errorf(codes.InvalidArgument,
				"TIDB_GROUP scope only accepts TIDB at SCALE_IN (NODE_COUNT, CPU) or SCALE_OUT (NODE_COUNT, CPU); unsupported keys: %s", strings.Join(bad, ","))
		}
	}

	// SHARED_POOL scope rejects components/keys outside the mapping
	// table — see design doc §Core Configuration Mapping.
	if scope == apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL {
		if bad := nonSharedPoolComponentsInSpecs(specs); len(bad) > 0 {
			return nil, status.Errorf(codes.InvalidArgument,
				"SHARED_POOL scope only accepts TIKV / TIKV_WORKER / TIDB_WORKER / COPROCESSOR_WORKER / SHARED_POOL / TIDB_GROUP components; unsupported: %s", strings.Join(bad, ","))
		}
		if bad := nonSharedPoolSpecKeysInSpecs(specs); len(bad) > 0 {
			return nil, status.Errorf(codes.InvalidArgument,
				"SHARED_POOL scope spec key outside mapping table; unsupported keys: %s", strings.Join(bad, ","))
		}
	}

	// SHARED_POOL/CLUSTER have no bundle-table row; skip the read so
	// a bundle-table error does not block a direct HPA/FeatureFlags
	// PATCH. See design doc §Inheritance Chains.
	var existing *repo.AutoscalePolicyBundleRow
	if scope == apiv1.AutoscalePolicyScope_SCOPE_DEFAULT ||
		scope == apiv1.AutoscalePolicyScope_SCOPE_TENANT {
		var err error
		existing, err = s.autoscalePolicyBundleRepo.GetBundle(ctx, scope, resolvedID)
		if err != nil {
			return nil, status.Errorf(codes.Internal, "load policy bundle: %v", err)
		}
	}
	var baseSpecs []*entity.AutoscalePolicySpec
	if existing != nil {
		baseSpecs = existing.Specs
	}
	merged := spec.MergeSpecs(baseSpecs, specs, spec.MergeModeMerge)
	hash := hashBundle(merged)

	// nil on the request preserves the existing row's value (PATCH
	// semantics). Runtime scopes have no stored directive.
	newApplyToAllBool := unwrapBoolValue(applyToAllSharedPools)
	if applyToAllSharedPools == nil && existing != nil {
		newApplyToAllBool = existing.ApplyToAllSharedPools
	}
	newApplyToAll := wrapBoolValue(newApplyToAllBool)

	row := &repo.AutoscalePolicyBundleRow{
		Scope:                 scope,
		SubjectID:             resolvedID,
		Specs:                 merged,
		ApplyToAllSharedPools: newApplyToAllBool,
		BundleHash:            hash,
	}
	// SHARED_POOL and CLUSTER store policy in HPA + ru_config.featureFlags,
	// not the bundle table. DEFAULT/TENANT use the bundle table.
	// See design doc §Inheritance Chains.
	if scope == apiv1.AutoscalePolicyScope_SCOPE_DEFAULT ||
		scope == apiv1.AutoscalePolicyScope_SCOPE_TENANT {
		if err := s.autoscalePolicyBundleRepo.UpsertBundle(ctx, row); err != nil {
			return nil, status.Errorf(codes.Internal, "upsert policy bundle: %v", err)
		}
	}

	// PATCH drives apply with the REQUEST delta, not the post-merge
	// bundle — otherwise a one-flag PATCH would re-apply every existing
	// flag value to every target.
	delta := spec.DeltaSpecs(specs)
	applyCtx := policyApplyContext{
		scope:                 scope,
		subjectID:             resolvedID,
		applyToAllSharedPools: newApplyToAll,
		requestID:             requestID,
		operation:             operationPatch,
	}
	targets, err := s.applyPolicyPatchToTargets(ctx, applyCtx, delta, merged)
	if err != nil {
		return nil, err
	}
	return &entity.AutoscalePolicyResponse{Targets: targets}, nil
}

// SyncAutoscalePolicy re-fans the canonical bundle to its targets
// without re-writing the bundle row. Mode controls fanout:
// REPLACE_ALL (default), KEYS_ONLY, DRY_RUN (Unimplemented; use
// PreviewAutoscalePolicy). applyToAllSharedPools is 3-state: nil
// preserves the stored value. See design doc §POST (Replace) / PATCH / Fanout.
func (s *serverImpl) SyncAutoscalePolicy(
	ctx context.Context,
	scope apiv1.AutoscalePolicyScope,
	subjectID string,
	mode apiv1.SyncAutoscalePolicyMode,
	keys []*entity.AutoscalePolicySpecKey,
	applyToAllSharedPools *wrapperspb.BoolValue,
	requestID string,
) (resp *entity.AutoscalePolicyResponse, err error) {
	// write_total + write_duration metric — same pattern as Create.
	startedAt := time.Now()
	defer func() {
		markPolicyWrite(scope, operationSync, applyToAllSharedPools,
			statusCodeToPolicyWriteResult(status.Code(err)), startedAt)
	}()

	if err := validateImplementedPolicyScope(scope); err != nil {
		return nil, err
	}
	resolvedID, err := resolveSubjectID(scope, subjectID)
	if err != nil {
		return nil, err
	}

	// SYNC is a fanout op — only meaningful at DEFAULT/TENANT.
	// SHARED_POOL/CLUSTER each address a single target. Reject
	// upfront (see design doc §Fanout Eligibility).
	if scope == apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL ||
		scope == apiv1.AutoscalePolicyScope_SCOPE_TIDB_GROUP {
		return nil, status.Errorf(codes.InvalidArgument,
			"SYNC is only valid at DEFAULT or TENANT scope (the two scopes with a fanout target set); got %s — single-target scopes use POST/PATCH directly",
			scope.String())
	}

	if err := validateApplyToAllSharedPoolsScope(scope, operationSync, applyToAllSharedPools); err != nil {
		return nil, err
	}
	// Default UNSPECIFIED → REPLACE_ALL. Allowlist rejects unknown
	// modes so they don't silently fall through to a full replace.
	if mode == apiv1.SyncAutoscalePolicyMode_SYNC_AUTOSCALE_POLICY_MODE_UNSPECIFIED {
		mode = apiv1.SyncAutoscalePolicyMode_SYNC_MODE_REPLACE_ALL
	}
	switch mode {
	case apiv1.SyncAutoscalePolicyMode_SYNC_MODE_REPLACE_ALL,
		apiv1.SyncAutoscalePolicyMode_SYNC_MODE_KEYS_ONLY,
		apiv1.SyncAutoscalePolicyMode_SYNC_MODE_DRY_RUN:
		// ok
	default:
		return nil, status.Errorf(codes.InvalidArgument,
			"unsupported SyncAutoscalePolicyMode %d (%s); allowed: REPLACE_ALL, KEYS_ONLY, DRY_RUN",
			mode, mode.String())
	}

	canonical, err := s.autoscalePolicyBundleRepo.GetBundle(ctx, scope, resolvedID)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "load canonical bundle: %v", err)
	}
	if canonical == nil {
		return &entity.AutoscalePolicyResponse{}, nil
	}

	// 3-state: request-level wins over stored; nil preserves stored.
	effectiveApplyToAll := applyToAllSharedPools
	if effectiveApplyToAll == nil {
		effectiveApplyToAll = wrapBoolValue(canonical.ApplyToAllSharedPools)
	}
	// TENANT SYNC always fans out — see design doc §Fanout Eligibility.
	if scope == apiv1.AutoscalePolicyScope_SCOPE_TENANT {
		effectiveApplyToAll = wrapperspb.Bool(true)
	}

	// Per-target failures flow through the targets list; only
	// target-discovery failures surface as an error (see design doc
	// §Inheritance Chains).
	targets, err := s.syncPolicyToTargets(ctx, canonical, mode, keys, effectiveApplyToAll)
	if err != nil {
		tilog.Warn(ctx, "policy: sync returned error",
			zap.String("scope", scope.String()),
			zap.String("subject_id", resolvedID),
			zap.String("mode", mode.String()),
			zap.Error(err))
		// Preserve typed gRPC codes (Unimplemented for DRY_RUN,
		// InvalidArgument for unknown modes) so callers keep the
		// signal.
		if _, ok := status.FromError(err); ok {
			return nil, err
		}
		return nil, status.Errorf(codes.Internal, "sync policy targets: %v", err)
	}
	return &entity.AutoscalePolicyResponse{Targets: targets}, nil
}

// ---------------------------------------------------------------------------
// Read RPCs
// ---------------------------------------------------------------------------

// GetAutoscalePolicy returns the canonical bundle at (scope, subject_id).
// Does NOT resolve across scopes — that's GetEffective's job. DEFAULT/TENANT
// read the bundle table; SHARED_POOL/CLUSTER read live runtime state.
// See design doc §Inheritance Chains.
func (s *serverImpl) GetAutoscalePolicy(
	ctx context.Context,
	scope apiv1.AutoscalePolicyScope,
	subjectID string,
) (*entity.AutoscalePolicySpecBundle, error) {
	if err := validateImplementedPolicyScope(scope); err != nil {
		return nil, err
	}
	resolvedID, err := resolveSubjectID(scope, subjectID)
	if err != nil {
		return nil, err
	}

	// SHARED_POOL/CLUSTER have no bundle row — read live runtime state.
	if scope == apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL ||
		scope == apiv1.AutoscalePolicyScope_SCOPE_TIDB_GROUP {
		specs, err := s.readRuntimeSpecs(ctx, scope, resolvedID)
		if err != nil {
			return nil, err
		}
		return &entity.AutoscalePolicySpecBundle{
			Scope:      scope,
			SubjectId:  resolvedID,
			Specs:      specs,
			BundleHash: hashBundle(specs),
		}, nil
	}

	row, err := s.autoscalePolicyBundleRepo.GetBundle(ctx, scope, resolvedID)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "get bundle: %v", err)
	}
	if row == nil {
		// No override at this scope — return an empty bundle
		// (resolvable to defaults in the effective view).
		return &entity.AutoscalePolicySpecBundle{
			Scope:     scope,
			SubjectId: resolvedID,
			Specs:     nil,
		}, nil
	}
	return &entity.AutoscalePolicySpecBundle{
		Scope:      row.Scope,
		SubjectId:  row.SubjectID,
		Specs:      row.Specs,
		BundleHash: row.BundleHash,
	}, nil
}

// GetEffectiveAutoscalePolicy resolves the bundle across the
// DEFAULT → TENANT → SHARED_POOL → CLUSTER priority chain.
// TIDB_GROUP scope itself reads live runtime state via readRuntimeSpecs.
// Per-key last-writer-wins; State=UNSPECIFIED on a higher scope =
// "follow the lower-priority scope". See design doc §Inheritance Chains.
func (s *serverImpl) GetEffectiveAutoscalePolicy(
	ctx context.Context,
	scope apiv1.AutoscalePolicyScope,
	subjectID string,
) (*entity.AutoscalePolicySpecBundle, error) {
	if err := validateImplementedPolicyScope(scope); err != nil {
		return nil, err
	}
	resolvedID, err := resolveSubjectID(scope, subjectID)
	if err != nil {
		return nil, err
	}
	// SHARED_POOL/CLUSTER read live runtime state; DEFAULT/TENANT read
	// the bundle table. See design doc §Inheritance Chains.
	var requestedSpecs []*entity.AutoscalePolicySpec
	if scope == apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL ||
		scope == apiv1.AutoscalePolicyScope_SCOPE_TIDB_GROUP {
		specs, err := s.readRuntimeSpecs(ctx, scope, resolvedID)
		if err != nil {
			return nil, status.Errorf(codes.Internal, "read runtime specs: %v", err)
		}
		requestedSpecs = specs
	} else {
		if row, err := s.autoscalePolicyBundleRepo.GetBundle(ctx, scope, resolvedID); err != nil {
			return nil, status.Errorf(codes.Internal, "load %s/%s: %v", scope, resolvedID, err)
		} else if row != nil {
			requestedSpecs = row.Specs
		}
	}
	return s.chainEffectiveSpecs(ctx, scope, resolvedID, requestedSpecs)
}

// chainEffectiveSpecs walks the priority chain (lower → higher scopes)
// and returns the merged effective bundle. `requestedSpecs` is supplied
// (not re-read from DB) so callers can compute "effective as if X" —
// used by PreviewAutoscalePolicy. CLUSTER is per-cluster (tidbGroupID),
// the TENANT level is looked up via tidb_groups.tenant_id. Higher-priority
// specs overwrite lower for the same key. See design doc §Inheritance Chains.
func (s *serverImpl) chainEffectiveSpecs(
	ctx context.Context,
	scope apiv1.AutoscalePolicyScope,
	resolvedID string,
	requestedSpecs []*entity.AutoscalePolicySpec,
) (*entity.AutoscalePolicySpecBundle, error) {
	type chainEntry struct {
		scope     apiv1.AutoscalePolicyScope
		subjectID string
	}

	var chainEntries []chainEntry
	switch scope {
	case apiv1.AutoscalePolicyScope_SCOPE_DEFAULT:
		chainEntries = []chainEntry{{scope: scope, subjectID: resolvedID}}
	case apiv1.AutoscalePolicyScope_SCOPE_TENANT:
		chainEntries = []chainEntry{
			{scope: apiv1.AutoscalePolicyScope_SCOPE_DEFAULT, subjectID: defaultAutoscalePolicySubjectID},
			{scope: apiv1.AutoscalePolicyScope_SCOPE_TENANT, subjectID: resolvedID},
		}
	case apiv1.AutoscalePolicyScope_SCOPE_TIDB_GROUP:
		// CLUSTER: DEFAULT → TENANT(actual tenant) → CLUSTER(tidbGroupID).
		// Tenant discovered via tidb_groups table.
		chainEntries = []chainEntry{
			{scope: apiv1.AutoscalePolicyScope_SCOPE_DEFAULT, subjectID: defaultAutoscalePolicySubjectID},
		}
		tenantID, lookupErr := s.tenantIDForTidbGroup(ctx, resolvedID)
		if lookupErr != nil {
			tilog.Warn(ctx, "policy: cannot resolve tenant for cluster; skipping tenant-level effective resolution",
				zap.String("tidbGroupId", resolvedID), zap.Error(lookupErr))
		}
		if tenantID != "" {
			chainEntries = append(chainEntries, chainEntry{
				scope: apiv1.AutoscalePolicyScope_SCOPE_TENANT, subjectID: tenantID,
			})
		}
		chainEntries = append(chainEntries, chainEntry{
			scope: apiv1.AutoscalePolicyScope_SCOPE_TIDB_GROUP, subjectID: resolvedID,
		})
	default:
		// SHARED_POOL: DEFAULT → TENANT(actual tenant) → SHARED_POOL(sharedPoolID).
		// Tenant discovered via tidb_clusters in this shared pool.
		chainEntries = []chainEntry{
			{scope: apiv1.AutoscalePolicyScope_SCOPE_DEFAULT, subjectID: defaultAutoscalePolicySubjectID},
		}
		tenantID, lookupErr := s.tenantIDForSharedPool(ctx, resolvedID)
		if lookupErr != nil {
			tilog.Warn(ctx, "policy: cannot resolve tenant for shared pool; skipping tenant-level effective resolution",
				zap.String("sharedPoolId", resolvedID), zap.Error(lookupErr))
		}
		if tenantID != "" {
			chainEntries = append(chainEntries, chainEntry{
				scope: apiv1.AutoscalePolicyScope_SCOPE_TENANT, subjectID: tenantID,
			})
		}
		chainEntries = append(chainEntries, chainEntry{
			scope: apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL, subjectID: resolvedID,
		})
	}

	merged := make(map[string]*entity.AutoscalePolicySpec)
	for _, ce := range chainEntries {
		var specs []*entity.AutoscalePolicySpec
		if ce.scope == scope {
			// Requested scope: use the caller's override (e.g. the
			// proposed bundle in PreviewAutoscalePolicy), not DB.
			specs = requestedSpecs
		} else {
			row, err := s.autoscalePolicyBundleRepo.GetBundle(ctx, ce.scope, ce.subjectID)
			if err != nil {
				return nil, status.Errorf(codes.Internal, "load chain scope %s: %v", ce.scope.String(), err)
			}
			if row != nil {
				specs = row.Specs
			}
		}
		for _, sp := range specs {
			if sp == nil {
				continue
			}
			key := fmt.Sprintf("%d|%d|%d|%d",
				sp.GetTargetType(), sp.GetDirection(), sp.GetCapability(), sp.GetTrigger())
			merged[key] = sp
		}
	}

	out := make([]*entity.AutoscalePolicySpec, 0, len(merged))
	for _, sp := range merged {
		out = append(out, sp)
	}
	sort.Slice(out, func(i, j int) bool { return specKeyString(out[i]) < specKeyString(out[j]) })

	return &entity.AutoscalePolicySpecBundle{
		Scope:      scope,
		SubjectId:  resolvedID,
		Specs:      out,
		BundleHash: hashBundle(out),
	}, nil
}

// specsEqualAsBundles returns true if a and b have the same
// (component, direction, capability, trigger) keys AND the same state
// for each key (nil specs ignored). Used by PreviewAutoscalePolicy to
// detect "canonical != effective" — priority-chain overrides on keys
// the requested scope also defines are a real semantic diff the
// preview must surface.
func specsEqualAsBundles(a, b []*entity.AutoscalePolicySpec) bool {
	if len(a) != len(b) {
		return false
	}
	am := make(map[string]apiv1.AutoscalePolicy_State, len(a))
	for _, s := range a {
		if s == nil {
			continue
		}
		k := fmt.Sprintf("%d|%d|%d|%d",
			s.GetTargetType(), s.GetDirection(), s.GetCapability(), s.GetTrigger())
		am[k] = s.GetState()
	}
	for _, s := range b {
		if s == nil {
			continue
		}
		k := fmt.Sprintf("%d|%d|%d|%d",
			s.GetTargetType(), s.GetDirection(), s.GetCapability(), s.GetTrigger())
		if v, ok := am[k]; !ok || v != s.GetState() {
			return false
		}
	}
	return true
}

func specKeyString(s *entity.AutoscalePolicySpec) string {
	return fmt.Sprintf("%d|%d|%d|%d",
		s.GetTargetType(), s.GetDirection(), s.GetCapability(), s.GetTrigger())
}

// ListAutoscalePolicy lists bundles by scope / subject_id prefix.
func (s *serverImpl) ListAutoscalePolicy(
	ctx context.Context,
	filter repo.AutoscalePolicyBundleFilter,
	pageSize uint32,
	pageToken string,
) ([]*entity.AutoscalePolicySpecBundle, string, error) {
	rows, next, err := s.autoscalePolicyBundleRepo.ListBundles(ctx, filter, pageSize, pageToken)
	if err != nil {
		return nil, "", status.Errorf(codes.Internal, "list bundles: %v", err)
	}
	bundles := make([]*entity.AutoscalePolicySpecBundle, 0, len(rows))
	for _, r := range rows {
		bundles = append(bundles, &entity.AutoscalePolicySpecBundle{
			Scope:      r.Scope,
			SubjectId:  r.SubjectID,
			Specs:      r.Specs,
			BundleHash: r.BundleHash,
		})
	}
	return bundles, next, nil
}

// ---------------------------------------------------------------------------
// Preview
// ---------------------------------------------------------------------------

// PreviewAutoscalePolicy is a dry-run. It computes the per-target
// effective before/after without writing. Returns one
// AutoscalePolicyPreviewTarget per affected scope (state-only diffs
// in MVP).
//
// Working: POST/PATCH at a single scope with no destructive replace.
// Not working (Unimplemented): PREVIEW_OP_SYNC, apply_to_all_shared_pools=true,
// destructive POST, and "canonical != effective" (any PATCH/POST at a
// scope whose priority chain has a higher-priority override). These
// four branches are detected upfront and return Unimplemented BEFORE
// any target discovery / diff computation, so callers that need the
// working subsets still get a real diff. See design doc §Preview.
// fanout / destructive / effective-state changes should
// monitor the related design doc §Open Questions follow-ups
// or use a different validation path (e.g. comparing
// GetAutoscalePolicy + GetEffectiveAutoscalePolicy in their
// own code) until a future PR wires this through.
func (s *serverImpl) PreviewAutoscalePolicy(
	ctx context.Context,
	scope apiv1.AutoscalePolicyScope,
	subjectID string,
	specs []*entity.AutoscalePolicySpec,
	operation apiv1.PreviewOperationType,
	applyToAllSharedPools *wrapperspb.BoolValue,
) ([]*entity.AutoscalePolicyPreviewTarget, error) {
	if err := validateImplementedPolicyScope(scope); err != nil {
		return nil, err
	}
	resolvedID, err := resolveSubjectID(scope, subjectID)
	if err != nil {
		return nil, err
	}

	// (1) PREVIEW_OP_SYNC: not implemented.
	if operation == apiv1.PreviewOperationType_PREVIEW_OP_SYNC {
		return nil, status.Errorf(codes.Unimplemented,
			"PREVIEW_OP_SYNC is not implemented in this build; SYNC's target-discovery + fanout semantics are deferred (see design doc §Open Questions)")
	}

	// (2) apply_to_all_shared_pools=true: per-target preview requires
	// walking each sharedpool's own rows + materialized policies,
	// which is deferred.
	if isApplyToAll(applyToAllSharedPools) {
		return nil, status.Errorf(codes.Unimplemented,
			"Preview with apply_to_all_shared_pools=true is not implemented in this build; per-target preview requires walking each sharedpool's own ru_configs.FeatureFlags + HPA, deferred")
	}

	// Load current canonical — needed for the destructive-replace
	// check and the per-field diff in the working path.
	current, err := s.GetAutoscalePolicy(ctx, scope, resolvedID)
	if err != nil {
		return nil, err
	}

	// (3) Destructive POST: request specs missing keys present in
	// current means the POST would remove them. diffSpecs can't
	// surface removals, so reject upfront. PATCH has no removals
	// by design, so it passes through.
	if operation == apiv1.PreviewOperationType_PREVIEW_OP_POST {
		if isDestructiveReplace(current.GetSpecs(), specs) {
			return nil, status.Errorf(codes.Unimplemented,
				"Preview with destructive POST (request specs missing keys present in current canonical) is not implemented in this build; diffSpecs cannot surface removals. Use a different validation path or wait for the per-key removals support to land")
		}
	}

	var mode spec.MergeMode
	switch operation {
	case apiv1.PreviewOperationType_PREVIEW_OP_POST:
		mode = spec.MergeModeReplace
	case apiv1.PreviewOperationType_PREVIEW_OP_PATCH:
		mode = spec.MergeModeMerge
	default:
		mode = spec.MergeModeReplace
	}

	proposedSpecs := spec.MergeSpecs(current.GetSpecs(), specs, mode)
	proposed := &entity.AutoscalePolicySpecBundle{
		Scope:      scope,
		SubjectId:  resolvedID,
		Specs:      proposedSpecs,
		BundleHash: hashBundle(proposedSpecs),
	}

	// Field-level diff (state-only for MVP). PATCH / non-destructive
	// POST are the only callers left here.
	diffs := diffSpecs(current.GetSpecs(), proposedSpecs)

	// (4) Canonical vs effective: if the priority chain has higher-
	// priority rows that override the canonical at the requested
	// scope, current_effective / proposed_effective differ from the
	// canonical bundles. Returning canonical as the effective would
	// silently misrepresent the runtime state, so reject upfront.
	// SCOPE_DEFAULT is always accepted (no higher-priority chain).
	currentEffective, err := s.chainEffectiveSpecs(ctx, scope, resolvedID, current.GetSpecs())
	if err != nil {
		return nil, err
	}
	proposedEffective, err := s.chainEffectiveSpecs(ctx, scope, resolvedID, proposed.GetSpecs())
	if err != nil {
		return nil, err
	}
	if !specsEqualAsBundles(currentEffective.GetSpecs(), current.GetSpecs()) ||
		!specsEqualAsBundles(proposedEffective.GetSpecs(), proposed.GetSpecs()) {
		return nil, status.Errorf(codes.Unimplemented,
			"Preview with canonical != effective (higher-priority DEFAULT/TENANT rows override the requested scope) is not implemented in this build; the per-target effective would diverge from the canonical bundle, and diffSpecs / field_diffs cannot represent the per-key state override. Use a different validation path (compare GetAutoscalePolicy + GetEffectiveAutoscalePolicy in your own code) or wait for per-target effective preview to land")
	}

	// Build the affected-target list. With the
	// apply_to_all_shared_pools=true branch rejected above,
	// this is always a single-target response.
	targets, err := s.previewPolicyTargets(ctx, scope, resolvedID, applyToAllSharedPools, current, proposed, diffs)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "preview targets: %v", err)
	}
	return targets, nil
}

// isDestructiveReplace returns true if `current` has any key not in
// `proposed` (POST would clear it). Used by PreviewAutoscalePolicy
// since diffSpecs only walks proposed. Key is the SpecKeyFor tuple —
// see pkg/autoscale/spec for SCALE_IN normalization (NODE_COUNT, CPU).
func isDestructiveReplace(current, proposed []*entity.AutoscalePolicySpec) bool {
	proposedKeys := make(map[string]struct{}, len(proposed))
	for _, s := range proposed {
		if s == nil {
			continue
		}
		k := spec.SpecKeyFor(s.GetTargetType(), s.GetDirection(), s.GetCapability(), s.GetTrigger())
		proposedKeys[k] = struct{}{}
	}
	for _, s := range current {
		if s == nil {
			continue
		}
		k := spec.SpecKeyFor(s.GetTargetType(), s.GetDirection(), s.GetCapability(), s.GetTrigger())
		if _, ok := proposedKeys[k]; !ok {
			return true
		}
	}
	return false
}

// diffSpecs returns per-key state changes between base and proposed.
// Both lists are pre-deduplicated (per-key, last wins). Keys in proposed
// but not in base appear as "added" (old = nil). Keys missing from
// proposed are not surfaced — PATCH never deletes a key.
func diffSpecs(base, proposed []*entity.AutoscalePolicySpec) []*entity.AutoscalePolicyFieldDiff {
	baseIdx := make(map[string]*entity.AutoscalePolicySpec)
	for _, s := range base {
		if s == nil {
			continue
		}
		baseIdx[specKeyString(s)] = s
	}
	var out []*entity.AutoscalePolicyFieldDiff
	for _, s := range proposed {
		if s == nil {
			continue
		}
		k := specKeyString(s)
		prev, ok := baseIdx[k]
		if ok && prev.GetState() == s.GetState() {
			continue
		}
		diff := &entity.AutoscalePolicyFieldDiff{
			TargetType: s.GetTargetType(),
			Direction:  s.GetDirection(),
			Capability: s.GetCapability(),
			Trigger:    s.GetTrigger(),
			FieldPath:  "state",
		}
		if ok {
			diff.OldValue = structpb.NewStringValue(prev.GetState().String())
		} else {
			diff.OldValue = nil
		}
		diff.NewValue = structpb.NewStringValue(s.GetState().String())
		out = append(out, diff)
	}
	return out
}

// ---------------------------------------------------------------------------
// Apply-to-targets (write paths)
// ---------------------------------------------------------------------------

// policyOperation identifies which RPC initiated the apply (so
// materialize can pick key-aware for PATCH, full-replace for POST/SYNC).
type policyOperation int

const (
	operationPatch policyOperation = iota
	operationCreateOrReplace
	operationSync
)

type policyApplyContext struct {
	scope                 apiv1.AutoscalePolicyScope
	subjectID             string
	applyToAllSharedPools *wrapperspb.BoolValue
	requestID             string
	operation             policyOperation
	// materializeMode overrides the default ru_configs.FeatureFlags write
	// strategy picked from `operation`. nil = "pick from operation"
	// (POST/SYNC → dense full-replace, PATCH → sparse key-aware). Callers
	// needing non-default (e.g. SYNC_MODE_KEYS_ONLY) set it explicitly.
	// Pointer so zero is distinguishable from explicit.
	materializeMode *materializeApplyMode
}

// applyPolicyPatchToTargets writes the FeatureFlags patch to the
// appropriate ru_configs row(s). At SHARED_POOL / TIDB_GROUP scope the
// `autoscale_policies` materialized view is dormant (design
// §Storage-point exclusivity) — HPA `spec.behavior` and
// `ru_configs.FeatureFlags` are the only sources of truth, so the
// legacy `materializeDeltaToAutoscalePolicies` SHARED_POOL/CLUSTER
// write path was removed. TIDB_GROUP writes for TIDB / TIPROXY go via
// writeTidbGroupHPABehavior → SetHPAScaleBehavior to per-group HPAs.
//
// `deltaSpecs` is request-only (UNSPECIFIED filtered) and drives the
// key-aware patch. `mergedSpecs` is the post-merge canonical bundle
// (used only by SHARED_POOL POST's full-replace path). For TENANT
// scope with apply_to_all_shared_pools=true, both fan out to every
// sharedpool of the tenant. Returns one AutoscalePolicyApplyTarget
// per write.
func (s *serverImpl) applyPolicyPatchToTargets(
	ctx context.Context,
	applyCtx policyApplyContext,
	deltaSpecs []*entity.AutoscalePolicySpec,
	mergedSpecs []*entity.AutoscalePolicySpec,
) ([]*entity.AutoscalePolicyApplyTarget, error) {
	// fanout_duration metric covers the full function for TENANT fanout
	// (target discovery + per-target write). Non-fanout branches ignore
	// this value. See design doc §Metrics.
	fanoutStartedAt := time.Now()
	patch, unrecognized := spec.SpecToFeatureFlags(deltaSpecs)
	// Dense block for full-replace writes (POST / SYNC REPLACE_ALL);
	// unused for PATCH (which uses the sparse patch).
	flagsBlock := spec.SpecsToFeatureFlagsBlock(deltaSpecs)
	if !patch.HasAnyChange() && len(deltaSpecs) == 0 {
		// Empty-delta: PATCH → SKIPPED. POST/SYNC: writes the
		// all-nil FeatureFlags block (ReplaceFeatureFlags on
		// the ru_configs row at SHARED_POOL scope). Full wipe
		// requires a not-yet-implemented ClearAutoscalePolicyBundle
		// RPC — see design doc §Open questions.
		if applyCtx.operation == operationPatch {
			return []*entity.AutoscalePolicyApplyTarget{{
				Scope:     applyCtx.scope,
				SubjectId: applyCtx.subjectID,
				Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SKIPPED,
				Message:   "no feature-flag change",
			}}, nil
		}
	}
	if err := s.guardDirectAutoscalePolicyMutation(ctx, applyCtx.scope, applyCtx.subjectID); err != nil {
		return nil, err
	}

	ruScope, ok := mapPolicyScopeToRuConfigScope(applyCtx.scope)
	switch {
	case !ok && applyCtx.scope == apiv1.AutoscalePolicyScope_SCOPE_DEFAULT:
		// DEFAULT has no ru_configs row in MVP. Canonical was stored;
		// return SKIPPED (DEFAULT rows are not fanned to sharedpools
		// in MVP — GetEffectiveAutoscalePolicy walks DEFAULT at resolve).
		return []*entity.AutoscalePolicyApplyTarget{{
			Scope:     applyCtx.scope,
			SubjectId: applyCtx.subjectID,
			Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SKIPPED,
			Message:   "DEFAULT scope has no ru_configs row in MVP; canonical stored",
		}}, nil
	case !ok && applyCtx.scope == apiv1.AutoscalePolicyScope_SCOPE_TIDB_GROUP:
		// CLUSTER is per-group, not per-sharedpool — apply the
		// TIDB/TIPROXY subset via writeTidbGroupHPABehavior. Bundle
		// table stores the user-written subset only; GetEffective
		// walks DEFAULT → TENANT → CLUSTER at read time.
		return s.writeTidbGroupHPABehavior(ctx, applyCtx, deltaSpecs, mergedSpecs)
	case !ok:
		// Defense in depth — entry-point validator should have caught this.
		return nil, status.Errorf(codes.InvalidArgument, "scope %s does not map to a ru_configs row in MVP", applyCtx.scope.String())
	}

	targets := []*entity.AutoscalePolicyApplyTarget{}
	if applyCtx.scope == apiv1.AutoscalePolicyScope_SCOPE_TENANT && isApplyToAll(applyCtx.applyToAllSharedPools) {
		// Fan out: each sharedpool of this tenant.
		ids, err := s.sharedPoolIDsForTenant(ctx, applyCtx.subjectID)
		if err != nil {
			return nil, err
		}
		for _, spID := range ids {
			// Best-effort: one failing write does not block the others;
			// caller reads per-target status. PATCH → sparse
			// PatchFeatureFlags; POST/SYNC → dense ReplaceFeatureFlags.
			targets = append(targets,
				s.writeRuConfigByOperation(ctx, spID, apiv1.RuConfig_SHARED_POOL, patch, flagsBlock, applyCtx, apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL),
			)
		}
		if len(targets) == 0 {
			// Tenant has no sharedpools — record a single skipped target.
			targets = append(targets, &entity.AutoscalePolicyApplyTarget{
				Scope:     applyCtx.scope,
				SubjectId: applyCtx.subjectID,
				Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SKIPPED,
				Message:   "tenant has no sharedpools",
			})
		}
		// Success-path audit log for fanout (design doc §Open Q #3).
		// Gated on `apply_to_all_shared_pools=true` — only fanout
		// operations produce meaningful target counts.
		var succ, fail int
		var failMsgs []string
		for _, tgt := range targets {
			// Per-target fanout_targets_total increment (synced/failed/skipped).
			markPolicyFanoutTarget(applyCtx.operation, tgt.GetStatus())
			switch tgt.GetStatus() {
			case apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SYNCED:
				succ++
			default:
				fail++
				if tgt.GetMessage() != "" {
					failMsgs = append(failMsgs, tgt.GetMessage())
				}
			}
		}
		operationStr := "unknown"
		switch applyCtx.operation {
		case operationCreateOrReplace:
			operationStr = "CreateOrReplace"
		case operationPatch:
			operationStr = "Patch"
		case operationSync:
			operationStr = "Sync"
		}
		tilog.Info(ctx, "autoscale policy fanout completed",
			zap.String("scope", applyCtx.scope.String()),
			zap.String("subject_id", applyCtx.subjectID),
			zap.String("request_id", applyCtx.requestID),
			zap.String("operation", operationStr),
			zap.Int("target_count", len(targets)),
			zap.Int("success_count", succ),
			zap.Int("failure_count", fail),
			zap.Strings("failures", failMsgs))
		// Wall-clock fanout_duration metric (target discovery + writes).
		observePolicyFanoutDuration(applyCtx.operation, fanoutStartedAt)
		return targets, nil
	}

	// apply_to_all_shared_pools only drives fanout at TENANT scope (the
	// branch above). For any other scope the flag is accepted but does
	// not fan out — the write applies to this subject only. Log it so an
	// operator who set the flag on e.g. a SHARED_POOL PATCH can see it
	// was intentionally scoped to the single subject rather than silently
	// dropped.
	if applyCtx.scope != apiv1.AutoscalePolicyScope_SCOPE_TENANT && isApplyToAll(applyCtx.applyToAllSharedPools) {
		tilog.Info(ctx, "autoscale policy: apply_to_all_shared_pools set on non-TENANT scope; applying to this subject only (no fanout)",
			zap.String("scope", applyCtx.scope.String()),
			zap.String("subject_id", applyCtx.subjectID),
			zap.String("request_id", applyCtx.requestID))
	}

	// TENANT non-fanout: the canonical bundle was already stored by the
	// caller (UpsertBundle in CreateOrReplace/Patch). Do NOT also write a
	// tenant ru_configs.FeatureFlags row — DEFAULT/TENANT policy writes
	// land only in the bundle table (no second authority). The
	// per-sharedpool FeatureFlags write happens on apply_to_all fanout
	// (above) or at CreateSharedPool freeze time.
	if applyCtx.scope == apiv1.AutoscalePolicyScope_SCOPE_TENANT {
		return []*entity.AutoscalePolicyApplyTarget{{
			Scope:     applyCtx.scope,
			SubjectId: applyCtx.subjectID,
			Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SKIPPED,
			Message:   "TENANT bundle stored; per-sharedpool write happens on apply_to_all fanout or CreateSharedPool",
		}}, nil
	}

	// SHARED_POOL (the only remaining scope here; DEFAULT/CLUSTER returned
	// above). Write both FeatureFlags-managed and HPA-managed portions:
	// FeatureFlags for TIKV/RCU/virtual components; HPA for worker components
	// (TIDB_WORKER / TIKV_WORKER / COPROCESSOR_WORKER) via SetHPAScaleBehavior.
	hpaTargets := s.writeSharedPoolWorkersHPABehavior(ctx, applyCtx, deltaSpecs, mergedSpecs)
	// Remaining unrecognized specs (non-worker, non-FeatureFlags) are
	// surfaced as SKIPPED targets (fallback for future edge cases).
	var remainingUnrecognized []*entity.AutoscalePolicySpec
	for _, sp := range unrecognized {
		ct := sp.GetTargetType()
		if ct == apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER ||
			ct == apiv1.AutoscaleTargetType_TARGET_TIKV_WORKER ||
			ct == apiv1.AutoscaleTargetType_TARGET_COPROCESSOR_WORKER {
			continue
		}
		remainingUnrecognized = append(remainingUnrecognized, sp)
	}
	hpaSkipped := hpaManagedSkipTargets(applyCtx, remainingUnrecognized)

	if !patch.HasAnyChange() {
		combined := make([]*entity.AutoscalePolicyApplyTarget, 0, len(hpaTargets)+len(hpaSkipped)+1)
		combined = append(combined, hpaTargets...)
		combined = append(combined, hpaSkipped...)
		if len(combined) > 0 {
			// POST/SYNC at SHARED_POOL scope: write the all-nil
			// FeatureFlags block to clear any prior ru_configs
			// overrides even when the request has only HPA-managed
			// specs (workers). PATCH preserves existing FeatureFlags
			// (merge semantics).
			if applyCtx.operation != operationPatch && hasAnySyncedTarget(hpaTargets) {
				combined = append(combined,
					s.writeRuConfigByOperation(ctx, applyCtx.subjectID, ruScope, patch, flagsBlock, applyCtx, applyCtx.scope))
			}
			return combined, nil
		}
		return []*entity.AutoscalePolicyApplyTarget{{
			Scope:     applyCtx.scope,
			SubjectId: applyCtx.subjectID,
			Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SKIPPED,
			Message:   "no feature-flag change and no HPA-managed change",
		}}, nil
	}

	targets = append(targets, s.writeRuConfigByOperation(ctx, applyCtx.subjectID, ruScope, patch, flagsBlock, applyCtx, applyCtx.scope))
	targets = append(targets, hpaTargets...)
	targets = append(targets, hpaSkipped...)
	return targets, nil
}

// hpaManagedSkipTargets converts HPA-only specs that cannot be represented by
// FeatureFlags into per-component SKIPPED targets. It is a fallback for specs
// not handled by writeSharedPoolWorkersHPABehavior (which handles the three
// worker component types at SHARED_POOL scope). This function is intentionally
// scoped to SHARED_POOL: TIDB_GROUP writes TIDB/TIPROXY through writeTidbGroupHPABehavior,
// while TENANT stores its canonical bundle and fanout targets are handled by
// their own scope.
// hasAnySyncedTarget returns true if any target has SYNCED status.
func hasAnySyncedTarget(targets []*entity.AutoscalePolicyApplyTarget) bool {
	for _, t := range targets {
		if t != nil && t.GetStatus() == apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SYNCED {
			return true
		}
	}
	return false
}

func hpaManagedSkipTargets(applyCtx policyApplyContext, unrecognized []*entity.AutoscalePolicySpec) []*entity.AutoscalePolicyApplyTarget {
	if applyCtx.scope != apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL {
		return nil
	}
	targets := make([]*entity.AutoscalePolicyApplyTarget, 0, len(unrecognized))
	for _, sp := range unrecognized {
		if sp == nil {
			continue
		}
		targets = append(targets, &entity.AutoscalePolicyApplyTarget{
			Scope:     applyCtx.scope,
			SubjectId: applyCtx.subjectID,
			Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SKIPPED,
			Message: fmt.Sprintf("delta spec for component %s is HPA-managed and out-of-band for this RPC; use legacy SetHPAScaleBehavior or trigger apply-component-hpa workflow",
				sp.GetTargetType()),
		})
	}
	return targets
}

// materializeApplyMode selects the ru_configs.FeatureFlags write
// strategy. The only currently-used value is KeyAware (sparse PATCH
// — patch only the flags the request specified, leave the rest).
// When applyCtx.materializeMode is nil the caller gets the default
// behavior, which is dense (full-replace) for POST/SYNC and sparse
// for PATCH. The unused FullReplace constant is documented in the
// switch in writeRuConfigByOperation.
type materializeApplyMode int

const (
	// materializeModeKeyAware: sparse write — patch only the flags the
	// request specified, leave the rest. Use for PATCH and SYNC KEYS_ONLY.
	materializeModeKeyAware materializeApplyMode = iota
)

func (s *serverImpl) writePatchToRuConfig(
	ctx context.Context,
	id string,
	scope apiv1.RuConfig_Scope,
	patch repo.FeatureFlagsPatch,
	originalScope apiv1.AutoscalePolicyScope,
) *entity.AutoscalePolicyApplyTarget {
	if err := s.ruConfigRepo.PatchFeatureFlags(ctx, id, scope, patch); err != nil {
		return &entity.AutoscalePolicyApplyTarget{
			Scope:     originalScope,
			SubjectId: id,
			Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_FAILED,
			Message:   err.Error(),
		}
	}
	return &entity.AutoscalePolicyApplyTarget{
		Scope:     originalScope,
		SubjectId: id,
		Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SYNCED,
		Message:   "patched ru_configs.feature_flags",
	}
}

// writeReplaceToRuConfig writes the dense FeatureFlags block via
// ReplaceFeatureFlags (REPLACE all 6 autoscale flags). POST /
// SYNC REPLACE_ALL counterpart to writePatchToRuConfig. nil flags
// = "reset to server default" on next priority resolution. The
// response message differs from writePatchToRuConfig so operators
// can tell which write method was used.
func (s *serverImpl) writeReplaceToRuConfig(
	ctx context.Context,
	id string,
	scope apiv1.RuConfig_Scope,
	flags *entity.RuConfigFeatureFlags,
	originalScope apiv1.AutoscalePolicyScope,
) *entity.AutoscalePolicyApplyTarget {
	if err := s.ruConfigRepo.ReplaceFeatureFlags(ctx, id, scope, flags); err != nil {
		return &entity.AutoscalePolicyApplyTarget{
			Scope:     originalScope,
			SubjectId: id,
			Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_FAILED,
			Message:   err.Error(),
		}
	}
	return &entity.AutoscalePolicyApplyTarget{
		Scope:     originalScope,
		SubjectId: id,
		Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SYNCED,
		Message:   "replaced ru_configs.feature_flags",
	}
}

// writeRuConfigByOperation dispatches to the right ru_config
// write method based on the operation's semantics. PATCH uses
// the sparse writePatchToRuConfig (per-field merge; nil =
// preserve). POST / SYNC REPLACE_ALL use the dense
// writeReplaceToRuConfig (full-replace; nil = clear). Both
// patch and flagsBlock are passed in to keep the caller's
// pre-computed values (so the test mock can assert on the
// exact patch / block that was used).
//
// Caller override (applyCtx.materializeMode == KeyAware) wins
// when set; this is how SYNC_MODE_KEYS_ONLY ends up as a sparse
// ru_config write even though `op == operationSync` (the
// default for operationSync is full-replace, which would clear
// non-mentioned flags — silent regression for flags the
// KEYS_ONLY caller didn't ask about).
func (s *serverImpl) writeRuConfigByOperation(
	ctx context.Context,
	id string,
	scope apiv1.RuConfig_Scope,
	patch repo.FeatureFlagsPatch,
	flagsBlock *entity.RuConfigFeatureFlags,
	applyCtx policyApplyContext,
	originalScope apiv1.AutoscalePolicyScope,
) *entity.AutoscalePolicyApplyTarget {
	if scope == apiv1.RuConfig_SHARED_POOL {
		if err := s.guardSharedPoolMutationByID(ctx, id, sharedPoolMutationBlockedWhilePaused); err != nil {
			return &entity.AutoscalePolicyApplyTarget{
				Scope: originalScope, SubjectId: id,
				Status: apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_FAILED, Message: err.Error(),
			}
		}
	}
	useSparse := applyCtx.materializeMode != nil && *applyCtx.materializeMode == materializeModeKeyAware
	switch {
	case useSparse:
		return s.writePatchToRuConfig(ctx, id, scope, patch, originalScope)
	case applyCtx.operation == operationCreateOrReplace, applyCtx.operation == operationSync:
		return s.writeReplaceToRuConfig(ctx, id, scope, flagsBlock, originalScope)
	case applyCtx.operation == operationPatch:
		return s.writePatchToRuConfig(ctx, id, scope, patch, originalScope)
	default:
		// Safe default: PATCH semantics (sparse).
		return s.writePatchToRuConfig(ctx, id, scope, patch, originalScope)
	}
}

func (s *serverImpl) guardDirectAutoscalePolicyMutation(ctx context.Context, scope apiv1.AutoscalePolicyScope, subjectID string) error {
	switch scope {
	case apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL:
		return s.guardSharedPoolMutationByID(ctx, subjectID, sharedPoolMutationBlockedWhilePaused)
	default:
		return nil
	}
}

// writeTidbGroupHPABehavior applies the TIDB_GROUP-scope user-written
// specs to the cluster's per-group-component HPA specs (TIDB /
// TIPROXY — per-group TIPROXY is accepted at TIDB_GROUP scope; pool-shared
// TIPROXY is naturally excluded by label selector mismatch; workers
// belong to SHARED_POOL scope). For each componentType in the request,
// calls SetHPAScaleBehavior once with per-direction flags derived from
// the specs (scaleUp from SCALE_OUT, scaleDown from SCALE_IN).
//
// Semantics per operation:
//
//   - operationPatch: pass only the request delta (deltaSpecs).
//     Directions absent from the request keep their existing HPA
//     value (SetHPAScaleBehavior nil=preserve).
//   - operationCreateOrReplace / operationSync: pass the full
//     user-written bundle (mergedSpecs). The bundle is the cluster's
//     own contribution — directions not in the bundle are NOT
//     cleared, they fall through to the chain at read time
//     (chainEffectiveSpecs composes DEFAULT → TENANT → CLUSTER).
//     This matches Path B: bundle stores user-written fields only,
//     read-time merge assembles the effective view.
//
// Non-TIDB / non-TIPROXY specs are filtered out (workers belong to
// SHARED_POOL scope). The cluster's tidbGroup is loaded to discover
// shared_pool_id (SetHPAScaleBehavior requires it).
//
// Returns one ApplyTarget per componentType written (or a single
// SKIPPED target if the bundle has no TIDB / TIPROXY specs).
func (s *serverImpl) writeTidbGroupHPABehavior(
	ctx context.Context,
	applyCtx policyApplyContext,
	deltaSpecs, mergedSpecs []*entity.AutoscalePolicySpec,
) ([]*entity.AutoscalePolicyApplyTarget, error) {
	if s.tidbGroupRepo == nil {
		return nil, status.Errorf(codes.Internal, "tidbGroupRepo not wired")
	}
	groupID := applyCtx.subjectID
	tg, err := s.tidbGroupRepo.Load(ctx, groupID, false)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "load tidb group %s: %v", groupID, err)
	}
	if tg == nil {
		return nil, status.Errorf(codes.NotFound, "tidb group %s not found", groupID)
	}
	spID := tg.GetSharedPoolId()
	if spID == "" {
		return nil, status.Errorf(codes.FailedPrecondition,
			"tidb group %s has no shared_pool_id; cannot resolve cluster HPA", groupID)
	}

	// Pick which specs to apply.
	//   PATCH → request delta only (per-key merge in HPA: directions
	//     absent from the request are preserved via nil in
	//     SetHPAScaleBehavior).
	//   POST / SYNC → full bundle (cluster's own contribution).
	//     Both POST and PATCH use nil-preserve for absent directions
	//     in the HPA write: a direction not in the source specs is
	//     passed as nil, SetHPAScaleBehavior skips it, and the HPA
	//     retains its current value. The priority chain
	//     (DEFAULT → TENANT → CLUSTER) resolves the effective
	//     policy at read time — Path B: read-time merge, no hard
	//     storage of inherited fields.
	sourceSpecs := deltaSpecs
	if applyCtx.operation == operationCreateOrReplace ||
		applyCtx.operation == operationSync {
		sourceSpecs = mergedSpecs
	}

	// Group specs by componentType, keeping only TIDB / TIPROXY.
	type compBucket struct {
		scaleUp, scaleDown *bool
	}
	byComp := map[apiv1.AutoscaleTargetType]*compBucket{}
	for _, sp := range sourceSpecs {
		ct := sp.GetTargetType()
		if ct != apiv1.AutoscaleTargetType_TARGET_TIDB && ct != apiv1.AutoscaleTargetType_TARGET_TIPROXY {
			continue
		}
		bucket, ok := byComp[ct]
		if !ok {
			bucket = &compBucket{}
			byComp[ct] = bucket
		}
		v := policyStateToBool(sp.GetState())
		switch sp.GetDirection() {
		case apiv1.AutoscalePolicy_SCALE_OUT:
			bucket.scaleUp = v
		case apiv1.AutoscalePolicy_SCALE_IN:
			bucket.scaleDown = v
		}
	}

	if len(byComp) == 0 {
		return []*entity.AutoscalePolicyApplyTarget{{
			Scope:     applyCtx.scope,
			SubjectId: applyCtx.subjectID,
			Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SKIPPED,
			Message:   "no TIDB or TIPROXY specs in CLUSTER bundle; cluster inherits from chain",
		}}, nil
	}

	targets := make([]*entity.AutoscalePolicyApplyTarget, 0, len(byComp))
	for ct, bucket := range byComp {
		// Translate back to ComponentType for the K8s-side
		// SetHPAScaleBehavior writer (it operates on real K8s
		// HPAs and so speaks the shared ComponentType enum).
		k8sCT := spec.AutoscaleTargetTypeToComponentType(ct)
		// SetHPAScaleBehavior's `groupId` parameter is overloaded
		// by componentType:
		//   TIDB    → tidbGroupID (HPA group label `db-<groupID>`)
		//   TIPROXY → tidbClusterID (HPA group label `db-<clusterID>`,
		//             created at generatedTiProxyGroup with comp.Id
		//             = tidbCluster.GetId())
		// Mirror apply_hpa.go's per-component identity model.
		idForSetHPA := groupID
		if k8sCT == apiv1.ComponentType_TIPROXY {
			idForSetHPA = tg.GetClusterId()
		}
		if err := s.SetHPAScaleBehavior(ctx, spID, idForSetHPA, k8sCT, bucket.scaleUp, bucket.scaleDown); err != nil {
			// A missing target HPA is a per-target no-op. Surface it as
			// SKIPPED rather than converting a topology mismatch into an
			// RPC-level failure. Other errors remain FAILED.
			if status.Code(err) == codes.Internal && strings.Contains(err.Error(), "has no hpa") {
				dirs := describeHPAWrite(bucket.scaleUp, bucket.scaleDown)
				targets = append(targets, &entity.AutoscalePolicyApplyTarget{
					Scope:     applyCtx.scope,
					SubjectId: applyCtx.subjectID,
					Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SKIPPED,
					Message: fmt.Sprintf("no target HPA for %s: %s (shared_pool_id=%s)",
						ct, dirs, spID),
				})
				continue
			}
			targets = append(targets, &entity.AutoscalePolicyApplyTarget{
				Scope:     applyCtx.scope,
				SubjectId: applyCtx.subjectID,
				Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_FAILED,
				Message:   fmt.Sprintf("set HPA behavior for %s: %v", ct, err),
			})
			continue
		}
		dirs := describeHPAWrite(bucket.scaleUp, bucket.scaleDown)
		targets = append(targets, &entity.AutoscalePolicyApplyTarget{
			Scope:     applyCtx.scope,
			SubjectId: applyCtx.subjectID,
			Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SYNCED,
			Message:   fmt.Sprintf("cluster HPA %s: %s (shared_pool_id=%s)", ct, dirs, spID),
		})
	}
	return targets, nil
}

// writeSharedPoolWorkersHPABehavior writes HPA behavior for worker
// components (TIDB_WORKER / TIKV_WORKER / COPROCESSOR_WORKER) at
// SHARED_POOL scope. Workers are pool-scoped (no groupId), so
// SetHPAScaleBehavior is called with groupId="". Mirrors
// writeTidbGroupHPABehavior at TIDB_GROUP scope.
//
// Semantics per operation:
//   - operationPatch: pass only the request delta (deltaSpecs).
//     Directions absent from the request keep their existing HPA value
//     (SetHPAScaleBehavior nil=preserve).
//   - operationCreateOrReplace / operationSync: pass the full
//     user-written bundle (mergedSpecs).
func (s *serverImpl) writeSharedPoolWorkersHPABehavior(
	ctx context.Context,
	applyCtx policyApplyContext,
	deltaSpecs, mergedSpecs []*entity.AutoscalePolicySpec,
) []*entity.AutoscalePolicyApplyTarget {
	sharedPoolID := applyCtx.subjectID

	// Pick which specs to apply.
	//   PATCH → request delta only (per-key merge in HPA: directions
	//     absent from the request are preserved via nil in
	//     SetHPAScaleBehavior).
	//   POST / SYNC → full bundle.
	sourceSpecs := deltaSpecs
	if applyCtx.operation == operationCreateOrReplace ||
		applyCtx.operation == operationSync {
		sourceSpecs = mergedSpecs
	}

	// Group specs by componentType, keeping only worker components
	// (TIDB_WORKER / TIKV_WORKER / COPROCESSOR_WORKER).
	type compBucket struct {
		scaleUp, scaleDown *bool
	}
	byComp := map[apiv1.AutoscaleTargetType]*compBucket{}
	for _, sp := range sourceSpecs {
		ct := sp.GetTargetType()
		if ct != apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER &&
			ct != apiv1.AutoscaleTargetType_TARGET_TIKV_WORKER &&
			ct != apiv1.AutoscaleTargetType_TARGET_COPROCESSOR_WORKER {
			continue
		}
		bucket, ok := byComp[ct]
		if !ok {
			bucket = &compBucket{}
			byComp[ct] = bucket
		}
		v := policyStateToBool(sp.GetState())
		switch sp.GetDirection() {
		case apiv1.AutoscalePolicy_SCALE_OUT:
			bucket.scaleUp = v
		case apiv1.AutoscalePolicy_SCALE_IN:
			bucket.scaleDown = v
		}
	}

	if len(byComp) == 0 {
		return nil
	}

	targets := make([]*entity.AutoscalePolicyApplyTarget, 0, len(byComp))
	for ct, bucket := range byComp {
		// Workers are pool-scoped (no groupId). Translate back to
		// ComponentType for the K8s-side SetHPAScaleBehavior writer
		// (it operates on real K8s HPAs and so speaks the shared
		// ComponentType enum).
		k8sCT := spec.AutoscaleTargetTypeToComponentType(ct)
		if err := s.SetHPAScaleBehavior(ctx, sharedPoolID, "", k8sCT, bucket.scaleUp, bucket.scaleDown); err != nil {
			// A missing target HPA is a per-target no-op. Surface it as
			// SKIPPED rather than converting a topology mismatch into an
			// RPC-level failure. Other errors remain FAILED.
			if status.Code(err) == codes.Internal && strings.Contains(err.Error(), "has no hpa") {
				dirs := describeHPAWrite(bucket.scaleUp, bucket.scaleDown)
				targets = append(targets, &entity.AutoscalePolicyApplyTarget{
					Scope:     applyCtx.scope,
					SubjectId: applyCtx.subjectID,
					Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SKIPPED,
					Message: fmt.Sprintf("no target HPA for %s: %s (shared_pool_id=%s)",
						ct, dirs, sharedPoolID),
				})
				continue
			}
			targets = append(targets, &entity.AutoscalePolicyApplyTarget{
				Scope:     applyCtx.scope,
				SubjectId: applyCtx.subjectID,
				Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_FAILED,
				Message:   fmt.Sprintf("set HPA behavior for %s: %v", ct, err),
			})
			continue
		}
		dirs := describeHPAWrite(bucket.scaleUp, bucket.scaleDown)
		targets = append(targets, &entity.AutoscalePolicyApplyTarget{
			Scope:     applyCtx.scope,
			SubjectId: applyCtx.subjectID,
			Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SYNCED,
			Message:   fmt.Sprintf("shared_pool HPA %s: %s (shared_pool_id=%s)", ct, dirs, sharedPoolID),
		})
	}
	return targets
}

// policyStateToBool translates an AutoscalePolicy_State into the
// *bool shape SetHPAScaleBehavior expects:
//   - ENABLED   → Bool(true)  (HPA engaged in this direction)
//   - DISABLED  → Bool(false) (HPA blocked in this direction)
//   - UNSPECIFIED → nil       (preserve existing HPA behavior)
//
// UNSPECIFIED happens when a request delta carries a sentinel-state
// spec (filtering leaves a residual entry whose State is the zero
// value). Treating it as nil-preserve is consistent with the
// "UNSPECIFIED carries no value" filter contract used throughout
// the bundle handlers.
func policyStateToBool(state apiv1.AutoscalePolicy_State) *bool {
	switch state {
	case apiv1.AutoscalePolicy_ENABLED:
		b := true
		return &b
	case apiv1.AutoscalePolicy_DISABLED:
		b := false
		return &b
	default:
		return nil
	}
}

func describeHPAWrite(scaleUp, scaleDown *bool) string {
	parts := []string{}
	if scaleUp != nil {
		if *scaleUp {
			parts = append(parts, "scale_up=enabled")
		} else {
			parts = append(parts, "scale_up=disabled")
		}
	}
	if scaleDown != nil {
		if *scaleDown {
			parts = append(parts, "scale_in=enabled")
		} else {
			parts = append(parts, "scale_in=disabled")
		}
	}
	if len(parts) == 0 {
		return "no-op (nil-preserve both directions)"
	}
	return strings.Join(parts, ", ")
}

// nonClusterComponentsInSpecs returns the deduplicated component-type
// names (in encounter order) that are NOT allowed at TIDB_GROUP scope.
// CLUSTER only governs per-group TIDB and TIPROXY HPA behavior on the
// write path; workers belong to SHARED_POOL scope (their HPA is
// per-sharedpool) and TIKV scale-in/out goes through
// ru_configs.FeatureFlags at SHARED_POOL scope. Used by
// the CreateOrReplace / Patch entry points to reject malformed CLUSTER
// requests with InvalidArgument before any DB write happens.
func nonClusterComponentsInSpecs(specs []*entity.AutoscalePolicySpec) []string {
	seen := map[apiv1.AutoscaleTargetType]bool{}
	var bad []string
	for _, sp := range specs {
		if sp == nil {
			continue
		}
		ct := sp.GetTargetType()
		if ct == apiv1.AutoscaleTargetType_TARGET_TIDB || ct == apiv1.AutoscaleTargetType_TARGET_TIPROXY {
			continue
		}
		if seen[ct] {
			continue
		}
		seen[ct] = true
		bad = append(bad, ct.String())
	}
	return bad
}

// clusterSpecKeyAllowlist is the (component, direction, capability,
// trigger) tuple set accepted at TIDB_GROUP scope. It mirrors exactly
// what hpaToSpecs emits on the readback path: per-group TIDB and
// TIPROXY rows at (NODE_COUNT, CPU) for BOTH SCALE_IN and SCALE_OUT.
// Anything outside this set is rejected at the entry point — see
// nonClusterSpecKeysInSpecs.
//
//   - Components: only TIDB and TIPROXY are written through this bundle
//     RPC at TIDB_GROUP scope. Workers (TIDB_WORKER / TIKV_WORKER /
//     COPROCESSOR_WORKER) belong to SHARED_POOL scope (their HPA is
//     per-sharedpool, not per-cluster); TIKV scale-in/out runs
//     through ru_configs.FeatureFlags at SHARED_POOL scope.
//   - Directions + (capability, trigger): the per-group HPA emits
//     both directions — ScaleUp.SelectPolicy and ScaleDown.SelectPolicy
//     — and SetHPAScaleBehavior writes both. The proto's
//     (capability, trigger) is a semantic label; at TIDB_GROUP scope
//     we use (NODE_COUNT, CPU) for BOTH directions, with `direction`
//     doing the directional split in the chain resolution.
//
// (Note: an earlier draft of this allowlist rejected SCALE_OUT for
// TIDB at TIDB_GROUP scope. That early draft was wrong — SetHPAScaleBehavior
// writes both ScaleUp and ScaleDown SelectPolicy for per-group TIDB HPAs,
// and writeTidbGroupHPABehavior accepts SCALE_OUT specs and maps them to
// scaleUp. The readback is the side that was missing the SCALE_OUT
// rows, not the
// write path. Fixing the readback here so the existing write path stays
// valid.)
//
// Keys are AutoscaleTargetType values (not the shared ComponentType),
// because the spec's tuple axis is the autoscaler-owned enum. Logical
// targets (SHARED_POOL / TIDB_GROUP) are not in this allowlist — they
// are FeatureFlags-backed, not HPA-backed.
var clusterSpecKeyAllowlist = map[apiv1.AutoscaleTargetType]map[apiv1.AutoscalePolicy_Direction]struct {
	Capability apiv1.AutoscalePolicy_Capability
	Trigger    apiv1.AutoscalePolicy_Trigger
}{
	apiv1.AutoscaleTargetType_TARGET_TIDB: {
		apiv1.AutoscalePolicy_SCALE_IN: {
			Capability: apiv1.AutoscalePolicy_NODE_COUNT,
			Trigger:    apiv1.AutoscalePolicy_CPU,
		},
		apiv1.AutoscalePolicy_SCALE_OUT: {
			Capability: apiv1.AutoscalePolicy_NODE_COUNT,
			Trigger:    apiv1.AutoscalePolicy_CPU,
		},
	},
	apiv1.AutoscaleTargetType_TARGET_TIPROXY: {
		apiv1.AutoscalePolicy_SCALE_IN: {
			Capability: apiv1.AutoscalePolicy_NODE_COUNT,
			Trigger:    apiv1.AutoscalePolicy_CPU,
		},
		apiv1.AutoscalePolicy_SCALE_OUT: {
			Capability: apiv1.AutoscalePolicy_NODE_COUNT,
			Trigger:    apiv1.AutoscalePolicy_CPU,
		},
	},
}

// nonClusterSpecKeysInSpecs returns the deduplicated
// "<component>:<direction>:<capability>:<trigger>" keys that are NOT
// allowed at TIDB_GROUP scope (per clusterSpecKeyAllowlist). Used by
// CreateOrReplace / Patch to reject malformed CLUSTER requests with
// InvalidArgument before any DB write happens. Companion to
// nonClusterComponentsInSpecs (which checks only the component axis;
// this one checks the full (component, direction, capability, trigger)
// tuple).
func nonClusterSpecKeysInSpecs(specs []*entity.AutoscalePolicySpec) []string {
	seen := map[string]bool{}
	var bad []string
	for _, sp := range specs {
		if sp == nil {
			continue
		}
		ct := sp.GetTargetType()
		dir := sp.GetDirection()
		capability := sp.GetCapability()
		trigger := sp.GetTrigger()
		if allowed, ok := clusterSpecKeyAllowlist[ct]; ok {
			if want, ok := allowed[dir]; ok && want.Capability == capability && want.Trigger == trigger {
				continue
			}
		}
		key := fmt.Sprintf("%s:%s:%s:%s", ct.String(), dir.String(), capability.String(), trigger.String())
		if seen[key] {
			continue
		}
		seen[key] = true
		bad = append(bad, key)
	}
	return bad
}

// sharedPoolSpecKeyAllowlist is the (component, direction, capability,
// trigger) tuple set accepted at SHARED_POOL scope. It mirrors exactly
// the design doc §Core Configuration Mapping Table:
//
//   - Workers (TIKV_WORKER / TIDB_WORKER / COPROCESSOR_WORKER):
//     SCALE_IN/SCALE_OUT at (NODE_COUNT, CPU) — HPA-managed.
//   - TIKV: SCALE_OUT at (NODE_COUNT, CPU), (DISK_SIZE, DISK),
//     (IOPS|THROUGHPUT, IO|DISK) — FeatureFlags-managed.
//   - SHARED_POOL (logical component): SCALE_OUT/SCALE_IN at
//     (NODE_SPEC, RCU) — enable_scale_shared_pool_by_ru.
//   - TIDB_GROUP (logical component): SCALE_OUT/SCALE_IN at
//     (NODE_SPEC, RCU) — enable_scale_tidb_group_by_ru.
//   - TIPROXY (pool-shared): SCALE_IN/SCALE_OUT at (NODE_COUNT, CPU) — HPA-backed (same as workers).
//
// Keys are AutoscaleTargetType values (not the shared ComponentType),
// because the spec's tuple axis is the autoscaler-owned enum.
var sharedPoolSpecKeyAllowlist = func() map[string]bool {
	ks := func(ct apiv1.AutoscaleTargetType, dir apiv1.AutoscalePolicy_Direction,
		cap apiv1.AutoscalePolicy_Capability, trig apiv1.AutoscalePolicy_Trigger) string {
		return fmt.Sprintf("%d|%d|%d|%d", ct, dir, cap, trig)
	}
	m := make(map[string]bool)

	// TIKV: SCALE_OUT × 3 capability/trigger axes.
	for _, ct := range []apiv1.AutoscalePolicy_Capability{
		apiv1.AutoscalePolicy_NODE_COUNT,
	} {
		m[ks(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, ct, apiv1.AutoscalePolicy_CPU)] = true
	}
	m[ks(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_DISK_SIZE, apiv1.AutoscalePolicy_DISK)] = true
	for _, ct := range []apiv1.AutoscalePolicy_Capability{
		apiv1.AutoscalePolicy_IOPS, apiv1.AutoscalePolicy_THROUGHPUT,
	} {
		for _, trig := range []apiv1.AutoscalePolicy_Trigger{
			apiv1.AutoscalePolicy_IO, apiv1.AutoscalePolicy_DISK,
		} {
			m[ks(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, ct, trig)] = true
		}
	}

	// TIPROXY (pool-shared): SCALE_IN/SCALE_OUT at (NODE_COUNT, CPU).
	for _, dir := range []apiv1.AutoscalePolicy_Direction{
		apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_SCALE_IN,
	} {
		m[ks(apiv1.AutoscaleTargetType_TARGET_TIPROXY, dir, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU)] = true
	}

	// Workers: SCALE_IN/SCALE_OUT at (NODE_COUNT, CPU).
	for _, comp := range []apiv1.AutoscaleTargetType{
		apiv1.AutoscaleTargetType_TARGET_TIKV_WORKER,
		apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER,
		apiv1.AutoscaleTargetType_TARGET_COPROCESSOR_WORKER,
	} {
		for _, dir := range []apiv1.AutoscalePolicy_Direction{
			apiv1.AutoscalePolicy_SCALE_OUT,
			apiv1.AutoscalePolicy_SCALE_IN,
		} {
			m[ks(comp, dir, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU)] = true
		}
	}

	// Virtual components: SHARED_POOL / TIDB_GROUP — SCALE_OUT/SCALE_IN
	// at (NODE_SPEC, RCU). Both directions map to the same bidirectional
	// FeatureFlags bit.
	for _, comp := range []apiv1.AutoscaleTargetType{
		apiv1.AutoscaleTargetType_SHARED_POOL,
		apiv1.AutoscaleTargetType_TIDB_GROUP,
	} {
		for _, dir := range []apiv1.AutoscalePolicy_Direction{
			apiv1.AutoscalePolicy_SCALE_OUT,
			apiv1.AutoscalePolicy_SCALE_IN,
		} {
			m[ks(comp, dir, apiv1.AutoscalePolicy_NODE_SPEC, apiv1.AutoscalePolicy_RCU)] = true
		}
	}

	return m
}()

// nonSharedPoolComponentsInSpecs returns the deduplicated component-type
// names (in encounter order) that are NOT allowed at SHARED_POOL scope.
// SHARED_POOL scope governs TIKV / TIKV_WORKER / TIDB_WORKER /
// COPROCESSOR_WORKER / SHARED_POOL (logical) / TIDB_GROUP (logical) /
// TIPROXY (pool-shared variant; per-group TIPROXY is CLUSTER-only).
// Used by the CreateOrReplace / Patch entry points to reject malformed
// SHARED_POOL requests with InvalidArgument before any DB write happens.
//
// Keys are AutoscaleTargetType values (not the shared ComponentType).
func nonSharedPoolComponentsInSpecs(specs []*entity.AutoscalePolicySpec) []string {
	allowed := map[apiv1.AutoscaleTargetType]bool{
		apiv1.AutoscaleTargetType_TARGET_TIKV:               true,
		apiv1.AutoscaleTargetType_TARGET_TIKV_WORKER:        true,
		apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER:        true,
		apiv1.AutoscaleTargetType_TARGET_COPROCESSOR_WORKER: true,
		apiv1.AutoscaleTargetType_SHARED_POOL:               true,
		apiv1.AutoscaleTargetType_TIDB_GROUP:                true,
		apiv1.AutoscaleTargetType_TARGET_TIPROXY:            true,
	}
	seen := map[apiv1.AutoscaleTargetType]bool{}
	var bad []string
	for _, sp := range specs {
		if sp == nil {
			continue
		}
		ct := sp.GetTargetType()
		if allowed[ct] {
			continue
		}
		if seen[ct] {
			continue
		}
		seen[ct] = true
		bad = append(bad, ct.String())
	}
	return bad
}

// nonSharedPoolSpecKeysInSpecs returns the deduplicated
// "<component>:<direction>:<capability>:<trigger>" keys that are NOT
// allowed at SHARED_POOL scope (per sharedPoolSpecKeyAllowlist).
// Companion to nonSharedPoolComponentsInSpecs (which checks only the
// component axis; this one checks the full tuple).
func nonSharedPoolSpecKeysInSpecs(specs []*entity.AutoscalePolicySpec) []string {
	seen := map[string]bool{}
	var bad []string
	ks := func(ct apiv1.AutoscaleTargetType, dir apiv1.AutoscalePolicy_Direction,
		cap apiv1.AutoscalePolicy_Capability, trig apiv1.AutoscalePolicy_Trigger) string {
		return fmt.Sprintf("%d|%d|%d|%d", ct, dir, cap, trig)
	}
	for _, sp := range specs {
		if sp == nil {
			continue
		}
		key := ks(sp.GetTargetType(), sp.GetDirection(), sp.GetCapability(), sp.GetTrigger())
		if sharedPoolSpecKeyAllowlist[key] {
			continue
		}
		if seen[key] {
			continue
		}
		seen[key] = true
		bad = append(bad, fmt.Sprintf("%s:%s:%s:%s",
			sp.GetTargetType().String(),
			sp.GetDirection().String(),
			sp.GetCapability().String(),
			sp.GetTrigger().String()))
	}
	return bad
}

// validateApplyToAllSharedPoolsScope enforces the proto contract that
// apply_to_all_shared_pools=true is only honored at TENANT scope AND only
// for PATCH / SYNC operations, per the design doc §Fanout Eligibility
// matrix. DEFAULT + apply_to_all=true is always rejected (would bypass
// every tenant override). SHARED_POOL / CLUSTER accept the flag as a
// no-op (single-target scopes have no fanout semantics). TENANT POST
// with apply_to_all=true is rejected because REPLACE fanout would wipe
// N sharedpools' FeatureFlags in one call.
// nil and false are always allowed.
//
// PreviewAutoscalePolicy has its own Unimplemented reject for this
// directive (per-target preview is deferred); it does not need this
// validator.
func validateApplyToAllSharedPoolsScope(scope apiv1.AutoscalePolicyScope, op policyOperation, b *wrapperspb.BoolValue) error {
	if !isApplyToAll(b) {
		return nil
	}
	// DEFAULT scope: fanout would bypass every tenant override — reject.
	if scope == apiv1.AutoscalePolicyScope_SCOPE_DEFAULT {
		return status.Errorf(codes.InvalidArgument,
			"apply_to_all_shared_pools=true is not valid at DEFAULT scope (would bypass every tenant override)")
	}
	// SHARED_POOL / CLUSTER: single-target scopes — accepted but no-op.
	if scope == apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL ||
		scope == apiv1.AutoscalePolicyScope_SCOPE_TIDB_GROUP {
		return nil
	}
	// TENANT POST with apply_to_all=true is rejected: REPLACE fanout
	// would wipe all sharedpools' FeatureFlags in one call (design
	// doc §Fanout Eligibility). PATCH and SYNC are supported.
	if op == operationCreateOrReplace {
		return status.Errorf(codes.InvalidArgument,
			"apply_to_all_shared_pools=true is not supported for POST (REPLACE); use PATCH (MERGE) or SYNC for fanout")
	}
	return nil
}

// validateBundleScopeComponentCompatibility validates component/scope
// pairs for bundle RPCs. Per-group TIPROXY is accepted at TIDB_GROUP scope
// (routed through writeTidbGroupHPABehavior, same as TIDB). Pool-shared
// TIPROXY at SHARED_POOL scope is accepted (label selectors handle
// topology filtering naturally — cross-topology writes result in SKIPPED
// with a clear message rather than an RPC-level error).
func validateBundleScopeComponentCompatibility(scope apiv1.AutoscalePolicyScope, specs []*entity.AutoscalePolicySpec) error {
	// No blanket component rejections at this layer — scope-specific
	// dispatch in applyPolicyPatchToTargets handles per-component routing.
	// Topology mismatches (e.g. per-group TIPROXY at SHARED_POOL) are
	// caught by HPA label selectors downstream and surfaced as SKIPPED.
	_ = scope
	_ = specs
	return nil
}

// syncPolicyToTargets re-fans the canonical bundle to its targets.
// For MVP, this is a thin wrapper around applyPolicyPatchToTargets with
// the bundle translated to a patch.
func (s *serverImpl) syncPolicyToTargets(
	ctx context.Context,
	canonical *repo.AutoscalePolicyBundleRow,
	mode apiv1.SyncAutoscalePolicyMode,
	keys []*entity.AutoscalePolicySpecKey,
	applyToAllSharedPools *wrapperspb.BoolValue,
) ([]*entity.AutoscalePolicyApplyTarget, error) {
	if mode == apiv1.SyncAutoscalePolicyMode_SYNC_MODE_DRY_RUN {
		// SYNC_MODE_DRY_RUN is parsed-and-allowed at the
		// SyncAutoscalePolicy entry point (see the explicit
		// allowlist above) but NOT implemented inside
		// syncPolicyToTargets. A previous reference implementation
		// returned a single SKIPPED target with a "dry-run"
		// message — which silently lied: the caller asked for a
		// per-target preview, got one fake target, and proceeded
		// as if the response was a real preview (it wasn't —
		// target discovery + diff computation were skipped).
		// Two options:
		//
		//   (a) wire this mode through the same target
		//       discovery / diff path used by
		//       PreviewAutoscalePolicy — invasive, deferred.
		//   (b) mark it unsupported and return Unimplemented
		//       with a clear "use Preview instead" message.
		//
		// Going with (b); the (a) implementation is tracked as
		// deferred work. The Unimplemented surfaces BEFORE any
		// target discovery happens, so a caller using DRY_RUN
		// gets a clear error rather than a silently-fake SKIPPED
		// response.
		// Pin test: TestSyncAutoscalePolicy_DryRunReturnsUnimplemented.
		return nil, status.Errorf(codes.Unimplemented,
			"SYNC_MODE_DRY_RUN is not implemented in this build; use PreviewAutoscalePolicy for per-target diff")
	}
	// Pick the spec slice we actually apply. The canonical bundle
	// row in DB stays the full bundle; only what we send to each
	// target's materialize step is filtered.
	applySpecs := canonical.Specs
	if mode == apiv1.SyncAutoscalePolicyMode_SYNC_MODE_KEYS_ONLY {
		// Sparse semantics: only the listed keys are re-applied
		// to each target. Only the matching flags are written to
		// each target's ru_configs.FeatureFlags; keys NOT in this
		// filtered slice stay untouched on each target (preserving
		// any existing FeatureFlags the SYNC KEYS_ONLY caller
		// didn't ask about).
		//
		// Comparison uses the natural key tuple
		// (component, direction, capability, trigger) computed by
		// spec.SpecKeyFor. The caller-supplied key list is
		// converted to a set of normalized strings once to keep
		// the per-spec check O(1).
		keep := make(map[string]struct{}, len(keys))
		for _, k := range keys {
			if k == nil {
				continue
			}
			keep[spec.SpecKeyFor(k.GetTargetType(), k.GetDirection(), k.GetCapability(), k.GetTrigger())] = struct{}{}
		}
		if len(keep) == 0 {
			// No valid keys supplied: nothing to do, surface as
			// a per-target SKIPPED with a clear message so the
			// caller knows their KEYS_ONLY request was a no-op
			// (not silently REPLACE_ALL, not a 0-target
			// success).
			return []*entity.AutoscalePolicyApplyTarget{{
				Scope:     canonical.Scope,
				SubjectId: canonical.SubjectID,
				Status:    apiv1.AutoscalePolicyApplyStatus_APPLY_STATUS_SKIPPED,
				Message:   "SYNC_MODE_KEYS_ONLY with empty keys list: nothing to apply",
			}}, nil
		}
		filtered := make([]*entity.AutoscalePolicySpec, 0, len(applySpecs))
		for _, s := range applySpecs {
			if s == nil {
				continue
			}
			k := spec.SpecKeyFor(s.GetTargetType(), s.GetDirection(), s.GetCapability(), s.GetTrigger())
			if _, ok := keep[k]; ok {
				filtered = append(filtered, s)
			}
		}
		applySpecs = filtered
	}
	// TENANT SYNC fans out to SHARED_POOL targets. Validate the
	// component against the target scope before discovering or writing
	// any fanout targets, so TIPROXY cannot be silently dropped by the
	// FeatureFlags mapper.
	if canonical.Scope == apiv1.AutoscalePolicyScope_SCOPE_TENANT &&
		isApplyToAll(applyToAllSharedPools) {
		if err := validateBundleScopeComponentCompatibility(apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL, applySpecs); err != nil {
			return nil, err
		}
	}
	// SYNC = REPLACE_ALL in MVP for non-KEYS_ONLY modes: the full
	// canonical bundle is the delta. The materialize helper writes
	//// the corresponding flags to each target's ru_configs.FeatureFlags
	// at SHARED_POOL scope.
	delta := spec.DeltaSpecs(applySpecs)
	applyCtx := policyApplyContext{
		scope:                 canonical.Scope,
		subjectID:             canonical.SubjectID,
		applyToAllSharedPools: applyToAllSharedPools,
		requestID:             "",
		operation:             operationSync,
	}
	// KEYS_ONLY is sparse: only the filtered keys are re-applied
	// to each target, so the materialize must NOT wipe other
	// rows (HPA behavior-derived rows, rows from prior operations) on the
	// target. Override the default full-replace strategy picked
	// from `operation == operationSync` above.
	if mode == apiv1.SyncAutoscalePolicyMode_SYNC_MODE_KEYS_ONLY {
		matMode := materializeModeKeyAware
		applyCtx.materializeMode = &matMode
	}
	return s.applyPolicyPatchToTargets(ctx, applyCtx, delta, applySpecs)
}

// previewPolicyTargets builds the per-target preview list. For MVP, only
// the (scope, subject_id) target is included unless apply_to_all
// is set at TENANT scope.
func (s *serverImpl) previewPolicyTargets(
	ctx context.Context,
	scope apiv1.AutoscalePolicyScope,
	subjectID string,
	applyToAllSharedPools *wrapperspb.BoolValue,
	current, proposed *entity.AutoscalePolicySpecBundle,
	diffs []*entity.AutoscalePolicyFieldDiff,
) ([]*entity.AutoscalePolicyPreviewTarget, error) {
	makeTarget := func(s2 apiv1.AutoscalePolicyScope, id string) *entity.AutoscalePolicyPreviewTarget {
		return &entity.AutoscalePolicyPreviewTarget{
			Scope:             s2,
			SubjectId:         id,
			CurrentCanonical:  current,
			CurrentEffective:  current, // MVP: canonical == effective
			ProposedCanonical: proposed,
			ProposedEffective: proposed,
			FieldDiffs:        diffs,
		}
	}

	if scope == apiv1.AutoscalePolicyScope_SCOPE_TENANT && isApplyToAll(applyToAllSharedPools) {
		ids, err := s.sharedPoolIDsForTenant(ctx, subjectID)
		if err != nil {
			return nil, err
		}
		if len(ids) == 0 {
			return []*entity.AutoscalePolicyPreviewTarget{
				makeTarget(scope, subjectID),
			}, nil
		}
		out := make([]*entity.AutoscalePolicyPreviewTarget, 0, len(ids))
		for _, spID := range ids {
			out = append(out, makeTarget(apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL, spID))
		}
		return out, nil
	}
	return []*entity.AutoscalePolicyPreviewTarget{makeTarget(scope, subjectID)}, nil
}

// sharedPoolIDsForTenant returns the IDs of sharedpools associated
// with the given tenant. Used by apply_to_all fanout (TENANT scope)
// and PreviewAutoscalePolicy.
//
// MVP limitation: the SharedPool schema does NOT persist TenantId
// (see pkg/storage/mysql/schema/schema.go — TenantId is a runtime-only
// entity field). The only "tenant → shared_pool" relationship is
// implicit through tidb_clusters: a tenant "owns" a shared pool once
// the first cluster is created there. Empty sharedpools (no clusters
// yet) are missed by this lookup.
//
// Caller implication: a fresh tenant that creates shared pools but
// has not yet created any clusters will see `apply_to_all=true` POST
// return without fanout. The user can re-POST once clusters exist,
// or a future migration can back-fill a shared_pool.tenant_id column.
//
// Implementation: walks tidb_clusters and dedups shared_pool_id per
// tenant. A tenant with no clusters returns an empty slice (no error).
func (s *serverImpl) sharedPoolIDsForTenant(ctx context.Context, tenantID string) ([]string, error) {
	if tenantID == "" {
		return nil, errors.New("tenantID is empty")
	}
	clusters, err := s.tidbClusterRepo.List(ctx, &repo.TiDBClusterListFilter{
		TenantIds: []string{tenantID},
		Deleted:   false,
	})
	if err != nil {
		return nil, fmt.Errorf("list tidb clusters for tenant %s: %w", tenantID, err)
	}
	seen := make(map[string]struct{}, len(clusters))
	out := make([]string, 0, len(clusters))
	for _, c := range clusters {
		spID := c.GetSharedPoolId()
		if spID == "" {
			continue
		}
		if _, ok := seen[spID]; ok {
			continue
		}
		seen[spID] = struct{}{}
		out = append(out, spID)
	}
	return out, nil
}

// tenantIDForSharedPool discovers the tenant that owns the given shared
// pool by looking up clusters in the shared pool and reading their
// tenant_id. This is the inverse of sharedPoolIDsForTenant.
//
// MVP limitation: empty shared pools without any tidb_clusters return
// ("", nil) — no error, but no tenant can be resolved. Callers should
// fall back to skipping the tenant-level resolution in the effective
// chain.
func (s *serverImpl) tenantIDForSharedPool(ctx context.Context, sharedPoolID string) (string, error) {
	if sharedPoolID == "" {
		return "", errors.New("sharedPoolID is empty")
	}
	clusters, err := s.tidbClusterRepo.List(ctx, &repo.TiDBClusterListFilter{
		SharedPoolIds: []string{sharedPoolID},
		Deleted:       false,
	})
	if err != nil {
		return "", fmt.Errorf("list tidb clusters for shared pool %s: %w", sharedPoolID, err)
	}
	for _, c := range clusters {
		if tid := c.GetTenantId(); tid != "" {
			return tid, nil // first cluster's tenant wins
		}
	}
	return "", nil // no clusters → no tenant
}

// readRuntimeSpecsForSharedPool / readRuntimeSpecsForCluster /
// readRuntimeSpecs live in autoscale_policy_runtime_specs.go.
// They read HPA spec.behavior + ru_configs.FeatureFlags directly,
// bypassing the dormant autoscale_policies table. This direct
// read is the source for BOTH the runtime decision path
// (EvaluateAutoscalePolicy) and the operator RPCs
// (GetAutoscalePolicy / GetEffectiveAutoscalePolicy).

// featureFlagsToSpecs converts ru_config.featureFlags to
// AutoscalePolicySpec entries. The function always emits the
// full set of 9 specs (3 TIKV flags + 1 shared IOPS+THROUGHPUT
// pair + 2 bidirectional RCU pairs + 1 TIPROXY SCALE_IN). Nil
// flag → DISABLED row. Non-nil flag → DISABLED if *flag is false,
// ENABLED if true.
//
// This handles both "FeatureFlags struct absent" (the pool has
// never been RCU-frozen — see create_sharedpool freeze step) and
// "individual flag unset" (partial FeatureFlags write) the same
// way: each unset key emits a DISABLED row. The caller — and any
// downstream chain resolution at GetEffectiveAutoscalePolicy —
// can then see every key explicitly with its resolved state.
//
// Bidirectional flags (RCU) emit both SCALE_OUT and SCALE_IN specs
// — the flag controls both directions. See design doc §Core
// Configuration Mapping Table rows 6–7.
func featureFlagsToSpecs(ff *apiv1.RuConfig_FeatureFlags) []*entity.AutoscalePolicySpec {
	// When ff is nil, dereferencing ff.X would panic. Capture each
	// flag pointer into a local var first (defaults to nil when
	// ff is nil) and pass the local to the helpers — keeps the
	// emission logic branch-free below.
	var (
		tikvCPU      *bool
		diskExpand   *bool
		iopsThrough  *bool
		sharedPoolRU *bool
		tidbGroupRU  *bool
		tiproxyIn    *bool
	)
	if ff != nil {
		tikvCPU = ff.EnableAutoscaleTikvCpuScaleOut
		diskExpand = ff.EnableAutoscaleDiskExpand
		iopsThrough = ff.EnableAutoscaleIopsThroughputExpand
		sharedPoolRU = ff.EnableScaleSharedPoolByRu
		tidbGroupRU = ff.EnableScaleTidbGroupByRu
		tiproxyIn = ff.EnableAutoscaleTiproxyScaleIn
	}

	var specs []*entity.AutoscalePolicySpec

	// emitBoolSpec appends a single spec keyed by (ct, dir, cap, trig).
	// Nil flag → DISABLED. Non-nil flag → DISABLED if *flag is false,
	// ENABLED if true. Always appends — callers do not branch on
	// nil; that way the bundle is complete regardless of which
	// flags were persisted.
	emitBoolSpec := func(
		ct apiv1.AutoscaleTargetType,
		dir apiv1.AutoscalePolicy_Direction,
		cap apiv1.AutoscalePolicy_Capability,
		trig apiv1.AutoscalePolicy_Trigger,
		flag *bool,
	) {
		state := apiv1.AutoscalePolicy_DISABLED
		if flag != nil && *flag {
			state = apiv1.AutoscalePolicy_ENABLED
		}
		specs = append(specs, &entity.AutoscalePolicySpec{
			TargetType: ct,
			Direction:  dir,
			Capability: cap,
			Trigger:    trig,
			State:      state,
		})
	}

	emitBoolSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT,
		apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, tikvCPU)
	emitBoolSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT,
		apiv1.AutoscalePolicy_DISK_SIZE, apiv1.AutoscalePolicy_DISK, diskExpand)

	// IOPS + THROUGHPUT share a single flag.
	stateIOPS := apiv1.AutoscalePolicy_DISABLED
	if iopsThrough != nil && *iopsThrough {
		stateIOPS = apiv1.AutoscalePolicy_ENABLED
	}
	specs = append(specs,
		&entity.AutoscalePolicySpec{
			TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
			Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
			Capability: apiv1.AutoscalePolicy_IOPS,
			Trigger:    apiv1.AutoscalePolicy_IO,
			State:      stateIOPS,
		},
		&entity.AutoscalePolicySpec{
			TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
			Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
			Capability: apiv1.AutoscalePolicy_THROUGHPUT,
			Trigger:    apiv1.AutoscalePolicy_IO,
			State:      stateIOPS,
		},
	)

	// RCU flags: SHARED_POOL and TIDB_GROUP are virtual
	// components (not real K8s components). Each flag is
	// bidirectional — one flag controls both SCALE_OUT and
	// SCALE_IN. See design doc §Core Configuration Mapping
	// Table rows 6–7.
	emitBoolSpec(apiv1.AutoscaleTargetType_SHARED_POOL, apiv1.AutoscalePolicy_SCALE_OUT,
		apiv1.AutoscalePolicy_NODE_SPEC, apiv1.AutoscalePolicy_RCU, sharedPoolRU)
	emitBoolSpec(apiv1.AutoscaleTargetType_SHARED_POOL, apiv1.AutoscalePolicy_SCALE_IN,
		apiv1.AutoscalePolicy_NODE_SPEC, apiv1.AutoscalePolicy_RCU, sharedPoolRU)
	emitBoolSpec(apiv1.AutoscaleTargetType_TIDB_GROUP, apiv1.AutoscalePolicy_SCALE_OUT,
		apiv1.AutoscalePolicy_NODE_SPEC, apiv1.AutoscalePolicy_RCU, tidbGroupRU)
	emitBoolSpec(apiv1.AutoscaleTargetType_TIDB_GROUP, apiv1.AutoscalePolicy_SCALE_IN,
		apiv1.AutoscalePolicy_NODE_SPEC, apiv1.AutoscalePolicy_RCU, tidbGroupRU)
	emitBoolSpec(apiv1.AutoscaleTargetType_TARGET_TIPROXY, apiv1.AutoscalePolicy_SCALE_IN,
		apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, tiproxyIn)
	return specs
}

// tenantIDForTidbGroup discovers the tenant that owns the given
// tidb group by loading the group row directly.
func (s *serverImpl) tenantIDForTidbGroup(ctx context.Context, groupID string) (string, error) {
	if groupID == "" {
		return "", errors.New("groupID is empty")
	}
	tg, err := s.tidbGroupRepo.Load(ctx, groupID, false)
	if err != nil {
		return "", fmt.Errorf("load tidb group %s: %w", groupID, err)
	}
	if tg == nil {
		return "", nil
	}
	return tg.GetTenantId(), nil
}

// mapPolicyScopeToRuConfigScope maps the v2 policy scope to the
// matching ru_configs scope. DEFAULT and CLUSTER return (_, false):
//
//   - DEFAULT scope has no corresponding ru_configs row in MVP (the
//     server default is read from YAML, not the DB). The v2 handler
//     still stores the canonical bundle in autoscale_policy_bundle_specs
//     and GetEffectiveAutoscalePolicy will read it on resolve; just no
//     ru_configs write happens. Callers should treat DEFAULT v2 POST
//     as "store the canonical; skip the apply step" — see
//     applyPolicyPatchToTargets for the SKIPPED target return.
//   - TIDB_GROUP scope has no ru_configs row (CLUSTER is per-group, not
//     per-sharedpool). The apply layer routes the user-written TIDB /
//     TIPROXY subset to `writeTidbGroupHPABehavior`, which writes the
//     per-group-component HPA directly; the bundle table also stores
//     the user-written subset. GetEffective walks DEFAULT → TENANT →
//     CLUSTER at read time and reads CLUSTER live state via
//     `readRuntimeSpecs`. Callers MUST special-case the
//     `(_, false) && SCOPE_TIDB_GROUP` branch — see
//     applyPolicyPatchToTargets.
//
// Callers MUST handle the (false) case explicitly.
func mapPolicyScopeToRuConfigScope(s apiv1.AutoscalePolicyScope) (apiv1.RuConfig_Scope, bool) {
	switch s {
	case apiv1.AutoscalePolicyScope_SCOPE_TENANT:
		return apiv1.RuConfig_TENANT, true
	case apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL:
		return apiv1.RuConfig_SHARED_POOL, true
	case apiv1.AutoscalePolicyScope_SCOPE_DEFAULT:
		// See godoc. Caller skips the ru_configs write.
		return apiv1.RuConfig_SCOPE_UNSPECIFIED, false
	default:
		// CLUSTER (and any unknown) has no ru_configs row — caller
		// dispatches to writeTidbGroupHPABehavior / returns
		// InvalidArgument for the unknown case.
		return apiv1.RuConfig_SCOPE_UNSPECIFIED, false
	}
}

// isApplyToAll returns true iff the request explicitly says
// apply_to_all_shared_pools=true. nil and false both mean "scoped
// only".
func isApplyToAll(b *wrapperspb.BoolValue) bool {
	return b != nil && b.GetValue()
}