package resolve

import (
	"context"
	"errors"
	"testing"

	"github.com/samber/lo"
	"github.com/stretchr/testify/require"
	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/storage/repo"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	v2 "k8s.io/api/autoscaling/v2"
)

// rcFF is a shorthand for building a RuConfigFeatureFlags with *bool fields.
// Use lo.ToPtr to construct each value so the test reads as "what flag is set
// to what". A nil argument to a *bool field means "tenant did not set this" —
// which is the proto3 optional semantics the design relies on.
func rcFF(scaleSharedPoolByRu, scaleTidbGroupByRu, tikvCpuScaleOut, diskExpand, iopsThroughputExpand, tiproxyScaleIn *bool) *entity.RuConfigFeatureFlags {
	return &entity.RuConfigFeatureFlags{
		EnableScaleSharedPoolByRu:           scaleSharedPoolByRu,
		EnableScaleTidbGroupByRu:            scaleTidbGroupByRu,
		EnableAutoscaleTikvCpuScaleOut:      tikvCpuScaleOut,
		EnableAutoscaleDiskExpand:           diskExpand,
		EnableAutoscaleIopsThroughputExpand: iopsThroughputExpand,
		EnableAutoscaleTiproxyScaleIn:       tiproxyScaleIn,
	}
}

// assertFlagsEqual compares two FeatureFlags pointer-by-pointer. A nil on
// either side must match nil on the other (no defensive .GetXxx() coercion —
// the resolution rule depends on the proto3 optional semantics).
func assertFlagsEqual(t *testing.T, want, got *entity.RuConfigFeatureFlags) {
	t.Helper()
	wantBools := []struct {
		name string
		val  *bool
	}{
		{"EnableScaleSharedPoolByRu", want.EnableScaleSharedPoolByRu},
		{"EnableScaleTidbGroupByRu", want.EnableScaleTidbGroupByRu},
		{"EnableAutoscaleTikvCpuScaleOut", want.EnableAutoscaleTikvCpuScaleOut},
		{"EnableAutoscaleDiskExpand", want.EnableAutoscaleDiskExpand},
		{"EnableAutoscaleIopsThroughputExpand", want.EnableAutoscaleIopsThroughputExpand},
		{"EnableAutoscaleTiproxyScaleIn", want.EnableAutoscaleTiproxyScaleIn},
	}
	gotBools := []struct {
		name string
		val  *bool
	}{
		{"EnableScaleSharedPoolByRu", got.EnableScaleSharedPoolByRu},
		{"EnableScaleTidbGroupByRu", got.EnableScaleTidbGroupByRu},
		{"EnableAutoscaleTikvCpuScaleOut", got.EnableAutoscaleTikvCpuScaleOut},
		{"EnableAutoscaleDiskExpand", got.EnableAutoscaleDiskExpand},
		{"EnableAutoscaleIopsThroughputExpand", got.EnableAutoscaleIopsThroughputExpand},
		{"EnableAutoscaleTiproxyScaleIn", got.EnableAutoscaleTiproxyScaleIn},
	}
	for i, w := range wantBools {
		g := gotBools[i]
		if w.name != g.name {
			t.Fatalf("internal test bug: name mismatch at index %d: %s vs %s", i, w.name, g.name)
		}
		switch {
		case w.val == nil && g.val == nil:
			// both nil — ok
		case w.val == nil && g.val != nil:
			t.Errorf("%s: want nil, got non-nil (%v)", w.name, *g.val)
		case w.val != nil && g.val == nil:
			t.Errorf("%s: want %v, got nil", w.name, *w.val)
		case *w.val != *g.val:
			t.Errorf("%s: want %v, got %v", w.name, *w.val, *g.val)
		}
	}
}

func TestOriginalScaleDownPolicy(t *testing.T) {
	tests := []struct {
		name       string
		component  apiv1.ComponentType
		wantPolicy v2.ScalingPolicySelect
	}{
		{name: "TIDB_WORKER — Max", component: apiv1.ComponentType_TIDB_WORKER, wantPolicy: v2.MaxChangePolicySelect},
		{name: "TIKV_WORKER — Max", component: apiv1.ComponentType_TIKV_WORKER, wantPolicy: v2.MaxChangePolicySelect},
		{name: "COPROCESSOR_WORKER — Min", component: apiv1.ComponentType_COPROCESSOR_WORKER, wantPolicy: v2.MinChangePolicySelect},
		{name: "TIDB — Min", component: apiv1.ComponentType_TIDB, wantPolicy: v2.MinChangePolicySelect},
		{name: "TIPROXY — Disabled", component: apiv1.ComponentType_TIPROXY, wantPolicy: v2.DisabledPolicySelect},
		{name: "TIKV — Disabled", component: apiv1.ComponentType_TIKV, wantPolicy: v2.DisabledPolicySelect},
		{name: "Unknown — Disabled", component: apiv1.ComponentType_COMPONENT_TYPE_UNSPECIFIED, wantPolicy: v2.DisabledPolicySelect},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			if got := OriginalScaleDownPolicy(tc.component); got != tc.wantPolicy {
				t.Errorf("got %v, want %v", got, tc.wantPolicy)
			}
		})
	}
}

func TestOriginalScaleUpPolicy(t *testing.T) {
	got := OriginalScaleUpPolicy()
	if got != v2.MaxChangePolicySelect {
		t.Errorf("OriginalScaleUpPolicy() = %v, want MaxChangePolicySelect", got)
	}
}

// TestSelectPolicyFromComponentState pins the (componentType, direction,
// enabled) → ScalingPolicySelect translation table that is the single
// source of truth shared by the bundle-aware reconcile path and the
// legacy write path (C1 fix). Cross-checked against the design doc
// §HPA Behavior Assembly Rule translation table.
func TestSelectPolicyFromComponentState(t *testing.T) {
	tests := []struct {
		name string
		// either uses compType+direction+enabled directly, or jumps
		// through the spec-driven helper to mirror pickSelectPolicyFromSpecs.
		component apiv1.ComponentType
		direction apiv1.AutoscalePolicy_Direction
		enabled   bool
		want      v2.ScalingPolicySelect
	}{
		// SCALE_OUT — baseline is MaxChange; ENABLED is a no-op,
		// DISABLED is the only state-changing row.
		{name: "SCALE_OUT ENABLED → MaxChange baseline",
			component: apiv1.ComponentType_TIDB, direction: apiv1.AutoscalePolicy_SCALE_OUT,
			enabled: true, want: v2.MaxChangePolicySelect},
		{name: "SCALE_OUT DISABLED → Disabled",
			component: apiv1.ComponentType_TIDB, direction: apiv1.AutoscalePolicy_SCALE_OUT,
			enabled: false, want: v2.DisabledPolicySelect},

		// SCALE_IN on NonDisabled-baseline components keeps the
		// baseline (TIDB/COPROCESSOR_WORKER=Min, TIKV_WORKER=Max).
		{name: "SCALE_IN ENABLED on TIDB baseline → Min",
			component: apiv1.ComponentType_TIDB, direction: apiv1.AutoscalePolicy_SCALE_IN,
			enabled: true, want: v2.MinChangePolicySelect},
		{name: "SCALE_IN ENABLED on TIKV_WORKER baseline → Max",
			component: apiv1.ComponentType_TIKV_WORKER, direction: apiv1.AutoscalePolicy_SCALE_IN,
			enabled: true, want: v2.MaxChangePolicySelect},
		{name: "SCALE_IN ENABLED on COPROCESSOR_WORKER baseline → Min",
			component: apiv1.ComponentType_COPROCESSOR_WORKER, direction: apiv1.AutoscalePolicy_SCALE_IN,
			enabled: true, want: v2.MinChangePolicySelect},

		// SCALE_IN ENABLED on Disabled-baseline components (TIPROXY,
		// TIKV) flips to MinChange. This is the C1 fix.
		{name: "SCALE_IN ENABLED on TIPROXY (Disabled baseline) → MinChange flip",
			component: apiv1.ComponentType_TIPROXY, direction: apiv1.AutoscalePolicy_SCALE_IN,
			enabled: true, want: v2.MinChangePolicySelect,
			// doc: design doc §HPA Behavior Assembly Rule — Disabled→MinChange flip.
		},
		{name: "SCALE_IN ENABLED on TIKV (Disabled baseline) → MinChange flip",
			component: apiv1.ComponentType_TIKV, direction: apiv1.AutoscalePolicy_SCALE_IN,
			enabled: true, want: v2.MinChangePolicySelect},

		// SCALE_IN DISABLED on every component disables scale-in —
		// the flip does NOT apply (Disabled intent is explicit).
		{name: "SCALE_IN DISABLED on TIDB → Disabled",
			component: apiv1.ComponentType_TIDB, direction: apiv1.AutoscalePolicy_SCALE_IN,
			enabled: false, want: v2.DisabledPolicySelect},
		{name: "SCALE_IN DISABLED on TIPROXY → Disabled",
			component: apiv1.ComponentType_TIPROXY, direction: apiv1.AutoscalePolicy_SCALE_IN,
			enabled: false, want: v2.DisabledPolicySelect},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			got := SelectPolicyFromComponentState(tc.component, tc.direction, tc.enabled)
			require.Equal(t, tc.want, got)
		})
	}
}

// specFor is a test helper that builds a minimal AutoscalePolicySpec
// for the (compType, direction, NODE_COUNT, CPU) tuple the helper
// ResolveHPABehaviorFromBundle looks up. Capability/Trigger default
// to NODE_COUNT/CPU; pass overrideCap/overrideTrig for the
// "non-CPU spec ignored" test case (T1.13).
func specFor(compType apiv1.AutoscaleTargetType, direction apiv1.AutoscalePolicy_Direction, state apiv1.AutoscalePolicy_State) *entity.AutoscalePolicySpec {
	return &entity.AutoscalePolicySpec{
		TargetType: compType,
		Direction:  direction,
		Capability: apiv1.AutoscalePolicy_NODE_COUNT,
		Trigger:    apiv1.AutoscalePolicy_CPU,
		State:      state,
	}
}

// specForDisk is a non-CPU spec (DISK_SIZE/DISK trigger) used to
// verify that ResolveHPABehaviorFromBundle does NOT translate
// FeatureFlag-driven specs into HPA Behavior (T1.13).
func specForDisk(compType apiv1.AutoscaleTargetType, direction apiv1.AutoscalePolicy_Direction, state apiv1.AutoscalePolicy_State) *entity.AutoscalePolicySpec {
	return &entity.AutoscalePolicySpec{
		TargetType: compType,
		Direction:  direction,
		Capability: apiv1.AutoscalePolicy_DISK_SIZE,
		Trigger:    apiv1.AutoscalePolicy_DISK,
		State:      state,
	}
}

func TestResolveHPABehaviorFromBundle(t *testing.T) {
	const tenantID = "tenant-A"

	tests := []struct {
		name         string
		defaultSpecs []*entity.AutoscalePolicySpec
		tenantSpecs  []*entity.AutoscalePolicySpec
		defaultErr   error
		tenantErr    error
		compType     apiv1.ComponentType
		wantUp       v2.ScalingPolicySelect
		wantDown     v2.ScalingPolicySelect
		wantErr      bool
		description  string
	}{
		{
			name: "T1.8 TENANT explicit SCALE_OUT DISABLED wins over baseline",
			tenantSpecs: []*entity.AutoscalePolicySpec{
				specFor(apiv1.AutoscaleTargetType_TARGET_TIDB, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_DISABLED),
			},
			compType: apiv1.ComponentType_TIDB,
			wantUp:   v2.DisabledPolicySelect,
			wantDown: v2.MinChangePolicySelect, // TIDB baseline
			description: "TENANT explicit spec wins regardless of DEFAULT. " +
				"SCALE_OUT DISABLED → DisabledPolicySelect. " +
				"SCALE_IN has no TENANT/DEFAULT spec → OriginalScaleDownPolicy(TIDB)=MinChange.",
		},
		{
			name: "T1.9 TENANT nil → DEFAULT SCALE_IN ENABLED wins",
			tenantSpecs: []*entity.AutoscalePolicySpec{
				// UNSPECIFIED is filtered out by MergeSpecs; this row carries no value.
				specFor(apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_STATE_UNSPECIFIED),
			},
			defaultSpecs: []*entity.AutoscalePolicySpec{
				specFor(apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER, apiv1.AutoscalePolicy_SCALE_IN, apiv1.AutoscalePolicy_ENABLED),
			},
			compType: apiv1.ComponentType_TIDB_WORKER,
			wantUp:   v2.MaxChangePolicySelect,
			wantDown: v2.MaxChangePolicySelect, // TIDB_WORKER baseline, ENABLED keeps baseline激进度
			description: "TENANT UNSPECIFIED is filtered; DEFAULT SCALE_IN ENABLED wins. " +
				"TIDB_WORKER baseline is MaxChange; ENABLED keeps it (no flip since baseline is not Disabled).",
		},
		{
			name:     "T1.10 both DEFAULT and TENANT nil/empty → baseline",
			compType: apiv1.ComponentType_TIKV_WORKER,
			wantUp:   v2.MaxChangePolicySelect,
			wantDown: v2.MaxChangePolicySelect, // TIKV_WORKER baseline
			description: "Neither row present → OriginalScale* baseline. " +
				"Pre-#2706 legacy behavior preserved.",
		},
		{
			name: "T1.11 SCALE_IN ENABLED on Disabled-baseline (TIPROXY) → MinChange flip",
			tenantSpecs: []*entity.AutoscalePolicySpec{
				specFor(apiv1.AutoscaleTargetType_TARGET_TIPROXY, apiv1.AutoscalePolicy_SCALE_IN, apiv1.AutoscalePolicy_ENABLED),
			},
			compType: apiv1.ComponentType_TIPROXY,
			wantUp:   v2.MaxChangePolicySelect,
			wantDown: v2.MinChangePolicySelect, // Disabled→MinChange flip
			description: "TIPROXY baseline is DisabledPolicySelect. " +
				"SCALE_IN ENABLED must enable scale-in (Disabled would defeat operator intent) " +
				"but does not force MaxChange — flips to MinChange.",
		},
		{
			name: "T1.12 SCALE_OUT DISABLED on MaxChange-baseline → DisabledPolicySelect",
			tenantSpecs: []*entity.AutoscalePolicySpec{
				specFor(apiv1.AutoscaleTargetType_TARGET_TIKV_WORKER, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_DISABLED),
			},
			compType: apiv1.ComponentType_TIKV_WORKER,
			wantUp:   v2.DisabledPolicySelect,
			wantDown: v2.MaxChangePolicySelect, // baseline, no SCALE_IN spec
			description: "SCALE_OUT DISABLED is a new capability — disables HPA scale-out " +
				"even though baseline is MaxChange.",
		},
		{
			name: "T1.13 non-CPU-direction spec (DISK_SIZE/DISK) is ignored",
			tenantSpecs: []*entity.AutoscalePolicySpec{
				// TIDB SCALE_OUT DISK_SIZE/DISK spec — this is not an HPA
				// Behavior axis (TIDB HPA only takes (NODE_COUNT, CPU)).
				// The helper must NOT translate it.
				specForDisk(apiv1.AutoscaleTargetType_TARGET_TIDB, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_ENABLED),
			},
			compType: apiv1.ComponentType_TIDB,
			wantUp:   v2.MaxChangePolicySelect, // baseline (no CPU spec)
			wantDown: v2.MinChangePolicySelect, // TIDB baseline
			description: "DISK_SIZE/DISK spec does not match the (NODE_COUNT, CPU) lookup key " +
				"so the helper falls back to OriginalScale* baseline.",
		},
		{
			name: "both bundle rows present, TENANT wins on conflicting key",
			tenantSpecs: []*entity.AutoscalePolicySpec{
				specFor(apiv1.AutoscaleTargetType_TARGET_COPROCESSOR_WORKER, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_DISABLED),
			},
			defaultSpecs: []*entity.AutoscalePolicySpec{
				specFor(apiv1.AutoscaleTargetType_TARGET_COPROCESSOR_WORKER, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_ENABLED),
			},
			compType: apiv1.ComponentType_COPROCESSOR_WORKER,
			wantUp:   v2.DisabledPolicySelect,
			wantDown: v2.MinChangePolicySelect, // COPROCESSOR_WORKER baseline
			description: "TENANT priority: even when DEFAULT has an ENABLED spec for the same " +
				"key, the TENANT DISABLED wins via MergeModeMerge.",
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			repoMock := mocks.NewAutoscalePolicyBundleRepo(t)
			repoMock.EXPECT().
				GetBundle(context.Background(), apiv1.AutoscalePolicyScope_SCOPE_DEFAULT, defaultBundleSubjectID).
				Return(&repo.AutoscalePolicyBundleRow{
					Scope:     apiv1.AutoscalePolicyScope_SCOPE_DEFAULT,
					SubjectID: defaultBundleSubjectID,
					Specs:     tc.defaultSpecs,
				}, tc.defaultErr).
				Maybe()
			repoMock.EXPECT().
				GetBundle(context.Background(), apiv1.AutoscalePolicyScope_SCOPE_TENANT, tenantID).
				Return(&repo.AutoscalePolicyBundleRow{
					Scope:     apiv1.AutoscalePolicyScope_SCOPE_TENANT,
					SubjectID: tenantID,
					Specs:     tc.tenantSpecs,
				}, tc.tenantErr).
				Maybe()

			up, down, err := ResolveHPABehaviorFromBundle(context.Background(), repoMock, tenantID, tc.compType)
			if tc.wantErr {
				require.Error(t, err)
				return
			}
			require.NoError(t, err)
			require.Equal(t, tc.wantUp, up, "%s — ScaleUp", tc.description)
			require.Equal(t, tc.wantDown, down, "%s — ScaleDown", tc.description)
		})
	}
}

func TestResolveHPABehaviorFromBundle_NilRepo(t *testing.T) {
	// No bundle repo → helper must return the per-component baseline
	// without touching the repo. This is the defensive path used by
	// legacy shared pools that haven't been wired through the v2
	// bundle table yet.
	up, down, err := ResolveHPABehaviorFromBundle(
		context.Background(), nil, "tenant-A", apiv1.ComponentType_TIDB)
	require.NoError(t, err)
	require.Equal(t, v2.MaxChangePolicySelect, up)
	require.Equal(t, v2.MinChangePolicySelect, down)
}

func TestResolveHPABehaviorFromBundle_EmptyTenantID(t *testing.T) {
	// Empty tenantID → defensive fallback to baseline. This handles
	// migration windows where sharedPool.TenantId might be unset.
	up, down, err := ResolveHPABehaviorFromBundle(
		context.Background(), nil, "", apiv1.ComponentType_TIKV_WORKER)
	require.NoError(t, err)
	require.Equal(t, v2.MaxChangePolicySelect, up)
	require.Equal(t, v2.MaxChangePolicySelect, down) // TIKV_WORKER baseline
}

func TestResolveHPABehaviorFromBundle_RepoError(t *testing.T) {
	// Repo errors must propagate so the caller (ApplyComponentHpaAction)
	// can log + fall back to baseline. The helper itself does NOT
	// silently fall back — that decision lives in the caller because
	// it is workflow-context-aware (retry semantics, metric emission).
	repoMock := mocks.NewAutoscalePolicyBundleRepo(t)
	repoMock.EXPECT().
		GetBundle(context.Background(), apiv1.AutoscalePolicyScope_SCOPE_DEFAULT, defaultBundleSubjectID).
		Return(nil, errors.New("db unavailable"))

	_, _, err := ResolveHPABehaviorFromBundle(
		context.Background(), repoMock, "tenant-A", apiv1.ComponentType_TIDB)
	require.Error(t, err)
	require.Contains(t, err.Error(), "load DEFAULT bundle row")
}

// TestPriorityResolveForSharedPoolFromBundle covers the new bundle-aware
// freeze-step entry point. Per-field rule: TENANT > DEFAULT > serverDefault
// (proto3 optional-bool: nil = fall through, non-nil = explicit value).
func TestPriorityResolveForSharedPoolFromBundle(t *testing.T) {
	const tenantID = "tenant-A"

	serverDef := &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			FeatureFlags: rcFF(lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(false)),
		},
	}

	tests := []struct {
		name         string
		defaultSpecs []*entity.AutoscalePolicySpec
		tenantSpecs  []*entity.AutoscalePolicySpec
		defaultRow   *repo.AutoscalePolicyBundleRow
		tenantRow    *repo.AutoscalePolicyBundleRow
		defaultErr   error
		tenantErr    error
		serverDef    *entity.RuConfig
		wantFlags    *entity.RuConfigFeatureFlags
		wantErr      bool
	}{
		{
			// DEF2.1: DEFAULT sets TIKV CPU scale-out ENABLED, TENANT
			// leaves it unset, server default says false. Expected:
			// TIKV CPU scale-out = true (DEFAULT wins over server
			// default), server default fills the other 4 flags.
			name: "DEF2.1 DEFAULT-only flag reaches resolved",
			defaultSpecs: []*entity.AutoscalePolicySpec{
				{
					TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
					Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
					Capability: apiv1.AutoscalePolicy_NODE_COUNT,
					Trigger:    apiv1.AutoscalePolicy_CPU,
					State:      apiv1.AutoscalePolicy_ENABLED,
				},
			},
			tenantSpecs: nil,
			serverDef:   serverDef,
			wantFlags:   rcFF(lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(true), lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(false)),
		},
		{
			// DEF2.2: Both DEFAULT and TENANT set TIKV CPU scale-out.
			// TENANT = DISABLED wins. The other 4 flags fall back to
			// server default (false).
			name: "DEF2.2 TENANT override of DEFAULT — TENANT wins",
			defaultSpecs: []*entity.AutoscalePolicySpec{
				{
					TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
					Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
					Capability: apiv1.AutoscalePolicy_NODE_COUNT,
					Trigger:    apiv1.AutoscalePolicy_CPU,
					State:      apiv1.AutoscalePolicy_ENABLED,
				},
			},
			tenantSpecs: []*entity.AutoscalePolicySpec{
				{
					TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
					Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
					Capability: apiv1.AutoscalePolicy_NODE_COUNT,
					Trigger:    apiv1.AutoscalePolicy_CPU,
					State:      apiv1.AutoscalePolicy_DISABLED,
				},
			},
			serverDef: serverDef,
			wantFlags: rcFF(lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(false)),
		},
		{
			// DEF2.3: DEFAULT sets disk-expand ENABLED, TENANT has
			// UNSPECIFIED-state specs only (filtered out by
			// SpecToFeatureFlags). Result: DEFAULT ENABLED wins for
			// disk-expand; other flags fall through to server default.
			name: "DEF2.3 TENANT nil-equivalent — DEFAULT set wins",
			defaultSpecs: []*entity.AutoscalePolicySpec{
				{
					TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
					Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
					Capability: apiv1.AutoscalePolicy_DISK_SIZE,
					Trigger:    apiv1.AutoscalePolicy_DISK,
					State:      apiv1.AutoscalePolicy_ENABLED,
				},
			},
			tenantSpecs: []*entity.AutoscalePolicySpec{
				{
					TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
					Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
					Capability: apiv1.AutoscalePolicy_NODE_COUNT,
					Trigger:    apiv1.AutoscalePolicy_CPU,
					State:      apiv1.AutoscalePolicy_STATE_UNSPECIFIED,
				},
			},
			serverDef: serverDef,
			wantFlags: rcFF(lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(true), lo.ToPtr(false), lo.ToPtr(false)),
		},
		{
			// DEF2.4: Both DEFAULT and TENANT empty / absent. All 5
			// flags fall back to server default (all false here).
			name:      "DEF2.4 both nil — server default wins",
			serverDef: serverDef,
			wantFlags: rcFF(lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(false), lo.ToPtr(false)),
		},
		{
			// All 5 flags exercised at once: a mix of per-tier
			// winners. DEFAULT sets disk-expand + iops, TENANT
			// overrides TIKV CPU to DISABLED and sets the two RCU
			// flags, server default fills tidb-group-by-ru.
			// Expected:
			//   shared_pool = TENANT ENABLED (TENANT)
			//   tidb_group  = TENANT ENABLED (TENANT)
			//   cpu         = TENANT DISABLED (TENANT)
			//   disk        = DEFAULT ENABLED (DEFAULT, TENANT unset)
			//   iops        = DEFAULT ENABLED (DEFAULT, TENANT unset)
			name: "DEF2.5 mixed priority across the 5 flags",
			defaultSpecs: []*entity.AutoscalePolicySpec{
				{
					TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
					Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
					Capability: apiv1.AutoscalePolicy_DISK_SIZE,
					Trigger:    apiv1.AutoscalePolicy_DISK,
					State:      apiv1.AutoscalePolicy_ENABLED,
				},
				{
					TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
					Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
					Capability: apiv1.AutoscalePolicy_IOPS,
					Trigger:    apiv1.AutoscalePolicy_IO,
					State:      apiv1.AutoscalePolicy_ENABLED,
				},
			},
			tenantSpecs: []*entity.AutoscalePolicySpec{
				{
					TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
					Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
					Capability: apiv1.AutoscalePolicy_NODE_COUNT,
					Trigger:    apiv1.AutoscalePolicy_CPU,
					State:      apiv1.AutoscalePolicy_DISABLED,
				},
				{
					TargetType: apiv1.AutoscaleTargetType_SHARED_POOL,
					Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
					Capability: apiv1.AutoscalePolicy_NODE_SPEC,
					Trigger:    apiv1.AutoscalePolicy_RCU,
					State:      apiv1.AutoscalePolicy_ENABLED,
				},
				{
					TargetType: apiv1.AutoscaleTargetType_TIDB_GROUP,
					Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
					Capability: apiv1.AutoscalePolicy_NODE_SPEC,
					Trigger:    apiv1.AutoscalePolicy_RCU,
					State:      apiv1.AutoscalePolicy_ENABLED,
				},
			},
			serverDef: serverDef,
			wantFlags: rcFF(lo.ToPtr(true), lo.ToPtr(true), lo.ToPtr(false), lo.ToPtr(true), lo.ToPtr(true), lo.ToPtr(false)),
		},
		{
			// nil server default → FeatureFlags block constructed
			// internally; all 5 pointers fall back to nil (no
			// server-default values to inherit).
			name:      "nil server default returns a non-nil FeatureFlags block with all-nil pointers",
			serverDef: nil,
			wantFlags: rcFF(nil, nil, nil, nil, nil, nil),
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			repoMock := mocks.NewAutoscalePolicyBundleRepo(t)
			defaultRow := tc.defaultRow
			if defaultRow == nil {
				defaultRow = &repo.AutoscalePolicyBundleRow{
					Scope:     apiv1.AutoscalePolicyScope_SCOPE_DEFAULT,
					SubjectID: defaultBundleSubjectID,
					Specs:     tc.defaultSpecs,
				}
			}
			tenantRow := tc.tenantRow
			if tenantRow == nil {
				tenantRow = &repo.AutoscalePolicyBundleRow{
					Scope:     apiv1.AutoscalePolicyScope_SCOPE_TENANT,
					SubjectID: tenantID,
					Specs:     tc.tenantSpecs,
				}
			}
			repoMock.EXPECT().
				GetBundle(context.Background(), apiv1.AutoscalePolicyScope_SCOPE_DEFAULT, defaultBundleSubjectID).
				Return(defaultRow, tc.defaultErr).
				Maybe()
			repoMock.EXPECT().
				GetBundle(context.Background(), apiv1.AutoscalePolicyScope_SCOPE_TENANT, tenantID).
				Return(tenantRow, tc.tenantErr).
				Maybe()

			got, err := PriorityResolveForSharedPoolFromBundle(
				context.Background(), repoMock, tenantID, tc.serverDef)
			if tc.wantErr {
				require.Error(t, err)
				return
			}
			require.NoError(t, err)
			require.NotNil(t, got)
			require.NotNil(t, got.FeatureFlags)
			assertFlagsEqual(t, tc.wantFlags, got.FeatureFlags)
		})
	}
}

// TestPriorityResolveForSharedPoolFromBundle_NilRepo confirms the
// defensive fallback when the bundle repo isn't wired: every field
// falls back to server default (which is itself nil for this test).
func TestPriorityResolveForSharedPoolFromBundle_NilRepo(t *testing.T) {
	got, err := PriorityResolveForSharedPoolFromBundle(
		context.Background(), nil, "tenant-A", nil)
	require.NoError(t, err)
	require.NotNil(t, got)
	require.NotNil(t, got.FeatureFlags)
	assertFlagsEqual(t, rcFF(nil, nil, nil, nil, nil, nil), got.FeatureFlags)
}

// TestPriorityResolveForSharedPoolFromBundle_EmptyTenantID confirms the
// defensive fallback when tenantID is empty: same all-nil behavior.
// Crucially, no repo read happens (matches the legacy
// PriorityResolveForSharedPool semantics on an empty tenantID).
func TestPriorityResolveForSharedPoolFromBundle_EmptyTenantID(t *testing.T) {
	// Use a strict mock that fails on any call. If the helper
	// mistakenly reads the bundle repo when tenantID=="", the mock
	// will fire and the test will fail.
	repoMock := mocks.NewAutoscalePolicyBundleRepo(t)

	got, err := PriorityResolveForSharedPoolFromBundle(
		context.Background(), repoMock, "", &entity.RuConfig{
			RuConfig: apiv1.RuConfig{
				FeatureFlags: rcFF(lo.ToPtr(true), lo.ToPtr(true), lo.ToPtr(true), lo.ToPtr(true), lo.ToPtr(true), lo.ToPtr(true)),
			},
		})
	require.NoError(t, err)
	require.NotNil(t, got)
	require.NotNil(t, got.FeatureFlags)
	assertFlagsEqual(t,
		rcFF(lo.ToPtr(true), lo.ToPtr(true), lo.ToPtr(true), lo.ToPtr(true), lo.ToPtr(true), lo.ToPtr(true)),
		got.FeatureFlags)
}

// TestPriorityResolveForSharedPoolFromBundle_RepoError ensures repo
// failures propagate (callers must surface Internal). The DEFAULT
// error is surfaced first because that read happens before the
// TENANT read.
func TestPriorityResolveForSharedPoolFromBundle_RepoError(t *testing.T) {
	repoMock := mocks.NewAutoscalePolicyBundleRepo(t)
	repoMock.EXPECT().
		GetBundle(context.Background(), apiv1.AutoscalePolicyScope_SCOPE_DEFAULT, defaultBundleSubjectID).
		Return(nil, errors.New("db unavailable"))

	got, err := PriorityResolveForSharedPoolFromBundle(
		context.Background(), repoMock, "tenant-A", nil)
	require.Error(t, err)
	require.Nil(t, got)
	require.Contains(t, err.Error(), "load DEFAULT bundle row")
}

// TestPriorityResolveForSharedPoolFromBundle_TenantRepoError ensures
// the TENANT read error also propagates after a successful DEFAULT
// read (the helper must not silently fall through if either read
// fails).
func TestPriorityResolveForSharedPoolFromBundle_TenantRepoError(t *testing.T) {
	repoMock := mocks.NewAutoscalePolicyBundleRepo(t)
	repoMock.EXPECT().
		GetBundle(context.Background(), apiv1.AutoscalePolicyScope_SCOPE_DEFAULT, defaultBundleSubjectID).
		Return(&repo.AutoscalePolicyBundleRow{
			Scope:     apiv1.AutoscalePolicyScope_SCOPE_DEFAULT,
			SubjectID: defaultBundleSubjectID,
			Specs:     nil,
		}, nil)
	repoMock.EXPECT().
		GetBundle(context.Background(), apiv1.AutoscalePolicyScope_SCOPE_TENANT, "tenant-A").
		Return(nil, errors.New("db unavailable"))

	got, err := PriorityResolveForSharedPoolFromBundle(
		context.Background(), repoMock, "tenant-A", nil)
	require.Error(t, err)
	require.Nil(t, got)
	require.Contains(t, err.Error(), "load TENANT bundle row for tenant tenant-A")
}