// Package spec implements the Spec → ru_configs.FeatureFlags translation
// for the unified autoscale policy configuration interface. See design
// doc §Core Configuration Mapping Table.
//
// State field is ENABLED / DISABLED / UNSPECIFIED. Only ENABLED and
// DISABLED are translated to a flag change; UNSPECIFIED is "no value
// for this key" and is filtered out before building the patch. The
// caller decides REPLACE (POST) vs MERGE (PATCH) — this helper only
// produces a patch where each non-nil pointer means "set this flag".
package spec

import (
	"fmt"
	"sort"

	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"
)

// ComponentTypeToAutoscaleTargetType translates a shared ComponentType
// to its autoscaler-side AutoscaleTargetType counterpart. The K8s
// components share the same identity but use the TARGET_ prefix in
// the new enum (the buf lint rule ENUM_VALUE_UNIQUE rejects
// same-name enum values in the same package, so TARGET_ is required).
//
// Returns AUTOSCALE_TARGET_TYPE_UNSPECIFIED for unknown inputs and
// for the logical SHARED_POOL / TIDB_GROUP, which are not real K8s
// components and therefore have no ComponentType counterpart.
//
// This is the single conversion point between the two enums. Callers
// that bridge from K8s-side data (e.g. live HPA reads) to the
// autoscaler-side AutoscalePolicySpec MUST go through this helper
// rather than casting integers — the values are NOT numerically
// identical on the wire (TARGET_TIKV=1 vs ComponentType_TIKV=1 is a
// coincidence; using int casts will silently break if either enum
// is renumbered).
func ComponentTypeToAutoscaleTargetType(c apiv1.ComponentType) apiv1.AutoscaleTargetType {
	switch c {
	case apiv1.ComponentType_TIKV:
		return apiv1.AutoscaleTargetType_TARGET_TIKV
	case apiv1.ComponentType_TIDB:
		return apiv1.AutoscaleTargetType_TARGET_TIDB
	case apiv1.ComponentType_TIPROXY:
		return apiv1.AutoscaleTargetType_TARGET_TIPROXY
	case apiv1.ComponentType_TIKV_WORKER:
		return apiv1.AutoscaleTargetType_TARGET_TIKV_WORKER
	case apiv1.ComponentType_COPROCESSOR_WORKER:
		return apiv1.AutoscaleTargetType_TARGET_COPROCESSOR_WORKER
	case apiv1.ComponentType_TIDB_WORKER:
		return apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER
	}
	return apiv1.AutoscaleTargetType_AUTOSCALE_TARGET_TYPE_UNSPECIFIED
}

// AutoscaleTargetTypeToComponentType is the reverse translator —
// used by writers that bridge from the autoscaler-side spec tuple
// (TargetType) back to the K8s-side ComponentType expected by the
// HPA-reconcile / SetHPAScaleBehavior path. Logical targets
// (SHARED_POOL / TIDB_GROUP) have no K8s counterpart and return
// COMPONENT_TYPE_UNSPECIFIED; callers MUST handle that case
// explicitly (those targets are FeatureFlags-backed, not HPA-backed).
func AutoscaleTargetTypeToComponentType(t apiv1.AutoscaleTargetType) apiv1.ComponentType {
	switch t {
	case apiv1.AutoscaleTargetType_TARGET_TIKV:
		return apiv1.ComponentType_TIKV
	case apiv1.AutoscaleTargetType_TARGET_TIDB:
		return apiv1.ComponentType_TIDB
	case apiv1.AutoscaleTargetType_TARGET_TIPROXY:
		return apiv1.ComponentType_TIPROXY
	case apiv1.AutoscaleTargetType_TARGET_TIKV_WORKER:
		return apiv1.ComponentType_TIKV_WORKER
	case apiv1.AutoscaleTargetType_TARGET_COPROCESSOR_WORKER:
		return apiv1.ComponentType_COPROCESSOR_WORKER
	case apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER:
		return apiv1.ComponentType_TIDB_WORKER
	}
	return apiv1.ComponentType_COMPONENT_TYPE_UNSPECIFIED
}

// SpecToFeatureFlags builds a FeatureFlagsPatch from a list of canonical
// AutoscalePolicySpec. UNSPECIFIED-state specs are dropped (they
// carry no value). Specs whose tuple doesn't map to a known flag
// are returned as unrecognized (caller can mark them SKIPPED). Full
// mapping table: see design doc §Core Configuration Mapping Table.
func SpecToFeatureFlags(specs []*entity.AutoscalePolicySpec) (repo.FeatureFlagsPatch, []*entity.AutoscalePolicySpec) {
	var patch repo.FeatureFlagsPatch
	var unrecognized []*entity.AutoscalePolicySpec

	for _, spec := range specs {
		if spec == nil {
			continue
		}
		if spec.GetState() == apiv1.AutoscalePolicy_STATE_UNSPECIFIED {
			// UNSPECIFIED = no value for this key. Skip it; do NOT
			// treat as "clear override". Clear-override is encoded
			// by the caller omitting the spec key entirely.
			continue
		}

		enabled := spec.GetState() == apiv1.AutoscalePolicy_ENABLED

		switch {
		// SCALE_OUT: TiKV CPU-driven node-count scale-out.
		case spec.GetDirection() == apiv1.AutoscalePolicy_SCALE_OUT &&
			spec.GetTargetType() == apiv1.AutoscaleTargetType_TARGET_TIKV &&
			spec.GetCapability() == apiv1.AutoscalePolicy_NODE_COUNT &&
			spec.GetTrigger() == apiv1.AutoscalePolicy_CPU:
			b := enabled
			patch.EnableAutoscaleTikvCpuScaleOut = &b

		// SCALE_OUT: TiKV disk-expand.
		case spec.GetDirection() == apiv1.AutoscalePolicy_SCALE_OUT &&
			spec.GetTargetType() == apiv1.AutoscaleTargetType_TARGET_TIKV &&
			spec.GetCapability() == apiv1.AutoscalePolicy_DISK_SIZE &&
			spec.GetTrigger() == apiv1.AutoscalePolicy_DISK:
			b := enabled
			patch.EnableAutoscaleDiskExpand = &b

		// SCALE_OUT: TiKV IOPS / throughput expand. Both capabilities
		// (iops, throughput) map to the same flag.
		case spec.GetDirection() == apiv1.AutoscalePolicy_SCALE_OUT &&
			spec.GetTargetType() == apiv1.AutoscaleTargetType_TARGET_TIKV &&
			(spec.GetCapability() == apiv1.AutoscalePolicy_IOPS ||
				spec.GetCapability() == apiv1.AutoscalePolicy_THROUGHPUT) &&
			(spec.GetTrigger() == apiv1.AutoscalePolicy_IO ||
				spec.GetTrigger() == apiv1.AutoscalePolicy_DISK):
			b := enabled
			patch.EnableAutoscaleIopsThroughputExpand = &b

		// SCALE_IN: TiProxy node-count scale-in. Trigger is ignored:
		// enable_autoscale_tiproxy_scale_in is the overall scale-in
		// gate, not a CPU-only switch. Capability stays NODE_COUNT
		// because the flag is horizontal (replica--) scale-in.
		case spec.GetDirection() == apiv1.AutoscalePolicy_SCALE_IN &&
			spec.GetTargetType() == apiv1.AutoscaleTargetType_TARGET_TIPROXY &&
			spec.GetCapability() == apiv1.AutoscalePolicy_NODE_COUNT:
			b := enabled
			patch.EnableAutoscaleTiproxyScaleIn = &b

		// SCALE: SHARED_POOL RCU quota adjustment. SHARED_POOL is a
		// logical target (not a real K8s component) introduced by the
		// autoscale policy surface. Both SCALE_OUT (RCU scale-up) and
		// SCALE_IN (RCU scale-down) map to the same flag — it is a
		// single bidirectional switch per design doc §Core Configuration
		// Mapping Table row 6.
		case spec.GetTargetType() == apiv1.AutoscaleTargetType_SHARED_POOL &&
			spec.GetCapability() == apiv1.AutoscalePolicy_NODE_SPEC &&
			spec.GetTrigger() == apiv1.AutoscalePolicy_RCU:
			b := enabled
			patch.EnableScaleSharedPoolByRu = &b

		// SCALE: TIDB_GROUP RCU quota adjustment. TIDB_GROUP is a
		// logical target (not a real K8s component); same
		// bidirectional-flag semantics as SHARED_POOL above.
		// Design doc §Core Configuration Mapping Table row 7.
		case spec.GetTargetType() == apiv1.AutoscaleTargetType_TIDB_GROUP &&
			spec.GetCapability() == apiv1.AutoscalePolicy_NODE_SPEC &&
			spec.GetTrigger() == apiv1.AutoscalePolicy_RCU:
			b := enabled
			patch.EnableScaleTidbGroupByRu = &b

		default:
			// SCALE_IN / SCALE_OUT for TARGET_TIDB / TARGET_TIDB_WORKER /
			// TARGET_TIKV_WORKER / TARGET_COPROCESSOR_WORKER and
			// TARGET_TIPROXY SCALE_OUT is HPA-controlled (via
			// `ResolveHPABehaviorFromBundle` → `reconcileExistingHPA` →
			// HPA `Behavior.Scale{Down,Up}.SelectPolicy`), not by
			// FeatureFlags. TARGET_TIPROXY SCALE_IN is FeatureFlags-
			// gated (`enable_autoscale_tiproxy_scale_in`). Callers
			// must use `SetHPAScaleBehavior` / `writeClusterHPABehavior`
			// for the HPA-backed directions.
			unrecognized = append(unrecognized, spec)
		}
	}

	return patch, unrecognized
}

// MergeSpecs merges a patch onto an existing canonical bundle at row
// level. REPLACE: wholesale replacement (caller passes base==nil).
// MERGE: per-key upsert with last-writer-wins.
//
// MVP PATCH contract: UNSPECIFIED-state specs are filtered out and do
// not affect base. PATCH cannot selectively remove keys — wipe a
// scope with POST specs:[].
//
// TODO(autoscale-policy): PATCH nil/UNSPECIFIED = no-op today (key is
// preserved, not deleted). When the DeleteAutoscalePolicy RPC lands
// (design doc §Open Question #14), wire per-key deletion through that
// path. The MergeModeMerge contract must then handle a delete sentinel
// (e.g. STATE_INHERIT or a separate delete_keys list). See
// domain/server/autoscale_policy_bundle.go PatchAutoscalePolicy.
func MergeSpecs(base, patch []*entity.AutoscalePolicySpec, mode MergeMode) []*entity.AutoscalePolicySpec {
	switch mode {
	case MergeModeReplace:
		// Wholesale replacement.
		return compactSpecs(patch)
	case MergeModeMerge:
		// Per-key merge. Spec key = (component_type, direction, capability, trigger).
		// UNSPECIFIED specs are dropped (they carry no value) —
		// same filtering as compactSpecs in the Replace path.
		keyIndex := make(map[string]*entity.AutoscalePolicySpec, len(base))
		for _, s := range base {
			if s == nil {
				continue
			}
			keyIndex[specKey(s)] = s
		}
		for _, patchSpec := range patch {
			if patchSpec == nil {
				continue
			}
			if patchSpec.GetState() == apiv1.AutoscalePolicy_STATE_UNSPECIFIED {
				continue
			}
			keyIndex[specKey(patchSpec)] = patchSpec
		}
		out := make([]*entity.AutoscalePolicySpec, 0, len(keyIndex))
		// Stable order: sort by key.
		out = append(out, sortedKeyIndex(keyIndex)...)
		return out
	default:
		return compactSpecs(patch)
	}
}

// MergeMode is the row-level merge strategy.
type MergeMode int

const (
	// MergeModeReplace wipes the base and applies the patch as the new
	// canonical row. Used by POST CreateOrReplaceAutoscalePolicy.
	MergeModeReplace MergeMode = iota
	// MergeModeMerge merges per-key with last-writer-wins. Used by PATCH.
	MergeModeMerge
)

// SpecKey returns the canonical key tuple
// "(component_type, direction, capability, trigger)" used for
// row-level merge in MergeSpecs and key-aware merge in the
// server-side materialize helper. Exported so callers outside
// the package can compute the same key (e.g. for matching an
// AutoscalePolicy row to a spec key).
//
// SpecKey is the caller-supplied (non-normalized) key. For the
// runtime-natural key that normalizes SCALE_IN to (NODE_COUNT, CPU),
// use SpecKeyFor. The materialize helper in
// domain/server/autoscale_policy_bundle.go uses SpecKeyFor for
// row keying (so the runtime cron's hardcoded query matches) but
// SpecKey is still the right key for the bundle-level merge where
// the bundle is caller-visible input.
func SpecKey(s *entity.AutoscalePolicySpec) string {
	return specKey(s)
}

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

// SpecKeyFor returns the runtime-natural key for an AutoscalePolicy,
// computed from the four tuple fields. SCALE_IN specs are normalized
// to (NODE_COUNT, CPU) so the key matches the only scale-in runtime
// caller's hardcoded query (RCU scale-down cron at
// domain/app/cron/rcu_scale_down.go:120-126). See
// docs/design/tenant-aware-autoscale-policy-redesign.md
// §Source-of-Truth Chain → Capability/Trigger Normalization.
//
// This is the key the server uses for spec identity and deduplication.
// The legacy `autoscale_policies` table is dormant for SHARED_POOL/CLUSTER
// (storage-point exclusivity); this key is used for the bundle-table merge
// and the caller-supplied spec key dedupe pass that collapses entries
// differing only in capability/trigger but normalizing to the same key.
func SpecKeyFor(
	target apiv1.AutoscaleTargetType,
	direction apiv1.AutoscalePolicy_Direction,
	capability apiv1.AutoscalePolicy_Capability,
	trigger apiv1.AutoscalePolicy_Trigger,
) string {
	if direction == apiv1.AutoscalePolicy_SCALE_IN {
		capability = apiv1.AutoscalePolicy_NODE_COUNT
		trigger = apiv1.AutoscalePolicy_CPU
	}
	return fmt.Sprintf("%d|%d|%d|%d", target, direction, capability, trigger)
}

func compactSpecs(in []*entity.AutoscalePolicySpec) []*entity.AutoscalePolicySpec {
	out := make([]*entity.AutoscalePolicySpec, 0, len(in))
	for _, s := range in {
		if s == nil {
			continue
		}
		if s.GetState() == apiv1.AutoscalePolicy_STATE_UNSPECIFIED {
			continue
		}
		out = append(out, s)
	}
	return out
}

// DeltaSpecs is the request-only representation used by PATCH to
// distinguish "the user explicitly set this key" from "this key was
// inherited from the existing canonical row". The delta is what
// drives the apply/fanout step — the canonical bundle row keeps
// using the post-merge `merged` representation.
//
// UNSPECIFIED-state specs are filtered out (they carry no value per
// MVP contract). The returned slice is suitable for:
//   - SpecToFeatureFlags (build a FeatureFlagsPatch that only
//     touches the keys the user explicitly set in the request).
//   - Per-target ru_configs.FeatureFlags write at SHARED_POOL scope
//     (patch only the keys the user explicitly set; preserve
//     everything else).
func DeltaSpecs(requestSpecs []*entity.AutoscalePolicySpec) []*entity.AutoscalePolicySpec {
	return compactSpecs(requestSpecs)
}

// SpecsToFeatureFlagsBlock builds a full RuConfigFeatureFlags
// block from a list of canonical specs. This is the DENSE
// counterpart to SpecToFeatureFlags (which builds a sparse
// FeatureFlagsPatch for PATCH). Every spec maps to one of the 6
// autoscale flags; flags NOT represented by any spec are
// returned as nil pointers (= "no override at this scope, fall
// through to default at read time"). Flags represented by a
// spec take the value (ENABLED → true, DISABLED → false) of
// the last spec that mentions them.
//
// Use this for POST CreateOrReplaceAutoscalePolicy and SYNC
// REPLACE_ALL semantics, where the canonical bundle is
// authoritative: every flag MUST be either an explicit value
// or nil (= no override). For POST `specs: []`, the result is
// an empty block with all 6 pointers nil, which is the MVP
// "wipe overrides" contract — ReplaceFeatureFlags will clear
// every autoscale flag at this scope.
//
// UNSPECIFIED-state specs are filtered out (they carry no
// value, same as SpecToFeatureFlags).
func SpecsToFeatureFlagsBlock(specs []*entity.AutoscalePolicySpec) *entity.RuConfigFeatureFlags {
	patch, _ := SpecToFeatureFlags(specs)
	return &entity.RuConfigFeatureFlags{
		EnableScaleSharedPoolByRu:           patch.EnableScaleSharedPoolByRu,
		EnableScaleTidbGroupByRu:            patch.EnableScaleTidbGroupByRu,
		EnableAutoscaleTikvCpuScaleOut:      patch.EnableAutoscaleTikvCpuScaleOut,
		EnableAutoscaleDiskExpand:           patch.EnableAutoscaleDiskExpand,
		EnableAutoscaleIopsThroughputExpand: patch.EnableAutoscaleIopsThroughputExpand,
		EnableAutoscaleTiproxyScaleIn:       patch.EnableAutoscaleTiproxyScaleIn,
	}
}

// sortedKeyIndex is a tiny helper to keep the merged output
// deterministic. It is not on the hot path; for the 6-flag MVP the
// result is at most ~20 entries.
func sortedKeyIndex(idx map[string]*entity.AutoscalePolicySpec) []*entity.AutoscalePolicySpec {
	out := make([]*entity.AutoscalePolicySpec, 0, len(idx))
	keys := make([]string, 0, len(idx))
	for k := range idx {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for _, k := range keys {
		out = append(out, idx[k])
	}
	return out
}