package server

// Direct-read adapter for the SHARED_POOL/CLUSTER autoscale policy
// read path. The 3 production read consumers
// (`EvaluateAutoscalePolicy`, `GetResolvedAutoscalePolicyBundle`,
// `ListAutoscalePolicies`) used to query the `autoscale_policies`
// table. Under the storage-point exclusivity invariant
// (docs/design/tenant-aware-autoscale-policy-redesign.md §Storage-point
// exclusivity), the SHARED_POOL/CLUSTER metadata lives in HPA
// `spec.behavior` + `ru_configs.FeatureFlags` only. This file
// composes those two sources (via the `readRuntimeSpecs` path,
// which reads HPA + FeatureFlags live on each call) and returns the
// same proto shapes the old readers used, so callers do not need to
// change.
//
// DEFAULT/TENANT scope is NOT touched here — bundle table reads
// continue to flow through `autoscalePolicyBundleRepo` from
// `domain/server/autoscale_policy_bundle.go`.

import (
	"context"

	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/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// directReadAutoscalePolicyMaterializedFrom marks the
// `MaterializedFrom` field on the ephemeral AutoscalePolicy rows
// we return from the direct-read path. The legacy
// `MATERIALIZED_FROM_UNSPECIFIED` (= 0) is the right fit: the row
// was not derived from any of the historical sources
// (HPA_BEHAVIOR / SHARED_POOL_OVERRIDE / CLUSTER_OVERRIDE), it
// was composed on the fly by the direct-read path. Consumers
// that branch on `MaterializedFrom` should treat the
// `UNSPECIFIED` value as "direct read, ephemeral".
const directReadAutoscalePolicyMaterializedFrom apiv1.AutoscalePolicy_MaterializedFrom = apiv1.AutoscalePolicy_MATERIALIZED_FROM_UNSPECIFIED

// evaluateAutoscalePolicyDirectRead returns the (allowed, reason)
// decision for a (sharedPool, component, direction, capability,
// trigger) tuple by direct-reading HPA + `ru_configs.FeatureFlags`
// for the shared pool and matching the requested key in memory.
//
// This replaces the legacy `listAutoscalePoliciesForSubject` +
// `findAutoscalePolicy` chain that read from the
// `autoscale_policies` table. The wire shape returned
// (`*entity.EvaluateAutoscalePolicyResponse`) is unchanged, so
// the 3 production callers
// (`checkScaleSharedPoolAutoscalePolicy`,
// `checkCapacityReservationAutoscalePolicy`, RCU cron) do not
// need to change.
func (s *serverImpl) evaluateAutoscalePolicyDirectRead(
	ctx context.Context,
	sharedPoolID string,
	componentType apiv1.ComponentType,
	direction apiv1.AutoscalePolicy_Direction,
	capability apiv1.AutoscalePolicy_Capability,
	trigger apiv1.AutoscalePolicy_Trigger,
) (*entity.EvaluateAutoscalePolicyResponse, error) {

	if sharedPoolID == "" {
		return &entity.EvaluateAutoscalePolicyResponse{
			Allowed: false,
			Reason:  "",
		}, nil
	}
	specs, err := s.readRuntimeSpecs(ctx, apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL, sharedPoolID)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "read runtime specs: %v", err)
	}
	// Two distinct no-policy outcomes (preserved from the legacy
	// `EvaluateAutoscalePolicy` contract):
	//   - no_policy_in_scope: the direct-read produced 0 specs
	//     for the subject. Implicit allow. Tenants that have not
	//     yet engaged with the policy surface (no FeatureFlags, no
	//     HPA behavior) are not denied scale activity.
	//   - missing_policy: the direct-read produced at least one
	//     spec, but none matches the requested (direction,
	//     capability, trigger) tuple. Deny.
	if len(specs) == 0 {
		autoscalePolicyResolveTotal.WithLabelValues(componentType.String(), capability.String(), trigger.String(), "no_policy_in_scope").Inc()
		return &entity.EvaluateAutoscalePolicyResponse{
			Allowed: true,
			Reason:  "no_policy_in_scope",
		}, nil
	}
	matched := findSpecByKeyInMemory(specs, componentType, direction, capability, trigger)
	if matched == nil {
		autoscalePolicyResolveTotal.WithLabelValues(componentType.String(), capability.String(), trigger.String(), "missing_policy").Inc()
		tilog.Info(ctx, "autoscale policy: missing_policy (deny — direct-read found specs but none matches the request key)",
			zap.String("sharedPoolId", sharedPoolID),
			zap.String("componentType", componentType.String()),
			zap.String("direction", direction.String()),
			zap.String("capability", capability.String()),
			zap.String("trigger", trigger.String()),
			zap.Int("availableSpecsCount", len(specs)))
		return &entity.EvaluateAutoscalePolicyResponse{
			Allowed: false,
			Reason:  "missing_policy",
		}, nil
	}
	resp := &entity.EvaluateAutoscalePolicyResponse{
		Allowed: matched.GetState() == apiv1.AutoscalePolicy_ENABLED,
		Reason:  specStateReason(matched),
		Policy:  specToAutoscalePolicy(sharedPoolID, matched),
	}
	autoscalePolicyResolveTotal.WithLabelValues(componentType.String(), capability.String(), trigger.String(), resp.GetReason()).Inc()
	return resp, nil
}

// getResolvedAutoscalePolicyBundleDirectRead returns the resolved
// per-sharedpool AutoscalePolicyBundle for `componentType` by
// composing HPA + `ru_configs.FeatureFlags` directly. The wire
// shape returned is identical to the legacy implementation that
// read from `autoscale_policies`, so the autoscale-engine
// consumer (which calls this via
// `pkg/autoscale/modules/common/controlplane_client.go::GetResolvedAutoscalePolicyBundle`)
// does not need to change.
//
// The `Version` field is set to the live spec count, matching
// the legacy `len(policies)` semantics.
func (s *serverImpl) getResolvedAutoscalePolicyBundleDirectRead(
	ctx context.Context,
	sharedPoolID string,
	componentType apiv1.ComponentType,
) (*entity.ResolvedAutoscalePolicyBundle, error) {
	specs, err := s.readRuntimeSpecs(ctx, apiv1.AutoscalePolicyScope_SCOPE_SHARED_POOL, sharedPoolID)
	if err != nil {
		return nil, err
	}
	policies := specsToAutoscalePolicies(sharedPoolID, componentType, specs)
	return &entity.ResolvedAutoscalePolicyBundle{
		SharedPoolId:  sharedPoolID,
		ComponentType: componentType,
		Version:       int64(len(policies)),
		Policies:      policies,
	}, nil
}

// findSpecByKeyInMemory returns the first spec in `specs` whose
// (component, direction, capability, trigger) key matches the
// query. SCALE_IN queries are normalized to
// `(NODE_COUNT, CPU)` per the SCALE_IN capability/trigger
// normalization rule (see SCALE_IN normalization & dedupe — the
// RCU cron uses the same hardcoded key at
// `domain/app/cron/rcu_scale_down.go:120-126`).
//
// componentType is part of the key — TIDB / TIKV_WORKER /
// TIDB_WORKER / COPROCESSOR_WORKER each have their own SCALE_IN
// spec, and the RCU cron evaluates every RCU-eligible component
// in `domain/server/rcu_scale_config.go`. Dropping component from
// the lookup key (e.g. by hardcoding TIKV) causes every non-TIKV
// SCALE_IN query to miss and be denied as `missing_policy` even
// when an HPA spec is present.
func findSpecByKeyInMemory(
	specs []*entity.AutoscalePolicySpec,
	componentType apiv1.ComponentType,
	direction apiv1.AutoscalePolicy_Direction,
	capability apiv1.AutoscalePolicy_Capability,
	trigger apiv1.AutoscalePolicy_Trigger,
) *entity.AutoscalePolicySpec {
	normalized := spec.SpecKeyFor(spec.ComponentTypeToAutoscaleTargetType(componentType), direction, capability, trigger)
	for _, s := range specs {
		if s == nil {
			continue
		}
		if spec.SpecKeyFor(s.GetTargetType(), s.GetDirection(), s.GetCapability(), s.GetTrigger()) == normalized {
			return s
		}
	}
	return nil
}

// specsToAutoscalePolicies converts a direct-read Spec list to
// the `*entity.AutoscalePolicy` rows that the legacy
// `GetResolvedAutoscalePolicyBundle` returned. We populate the
// 5 spec-derived fields + `SubjectType=SHARED_POOL`,
// `SubjectId=spID` so the consumer (autoscale-engine) sees the
// same row shape it always has. The metadata fields
// (`MaterializedFrom`, `LastModifiedSource`, etc.) are not
// populated — the engine does not read them.
func specsToAutoscalePolicies(spID string, componentType apiv1.ComponentType, specs []*entity.AutoscalePolicySpec) []*entity.AutoscalePolicy {
	out := make([]*entity.AutoscalePolicy, 0, len(specs))
	for _, s := range specs {
		if s == nil {
			continue
		}
		// Translate the spec's TargetType to ComponentType for
		// comparison with the K8s-side filter arg, and for the
		// AutoscalePolicy row's ComponentType field below.
		// Specs keyed by SHARED_POOL / TIDB_GROUP are
		// FeatureFlags-backed — they have no K8s counterpart and
		// would never appear in this direct-read path
		// (autoscaler-engine only consumes TIDB/TIPROXY rows at
		// CLUSTER scope).
		k8sCT := spec.AutoscaleTargetTypeToComponentType(s.GetTargetType())
		if componentType != apiv1.ComponentType_COMPONENT_TYPE_UNSPECIFIED && k8sCT != componentType {
			continue
		}
		out = append(out, specToAutoscalePolicy(spID, s))
	}
	return out
}

func specToAutoscalePolicy(spID string, s *entity.AutoscalePolicySpec) *entity.AutoscalePolicy {
	return &entity.AutoscalePolicy{
		SubjectType:      apiv1.AutoscalePolicy_SHARED_POOL,
		SubjectId:        spID,
		ComponentType:    spec.AutoscaleTargetTypeToComponentType(s.GetTargetType()),
		Direction:        s.GetDirection(),
		Capability:       s.GetCapability(),
		Trigger:          s.GetTrigger(),
		State:            s.GetState(),
		MaterializedFrom: directReadAutoscalePolicyMaterializedFrom,
	}
}

// specStateReason maps an AutoscalePolicySpec. State value to the reason
// string used in EvaluateAutoscalePolicy responses under the direct-read path.
func specStateReason(s *entity.AutoscalePolicySpec) string {
	switch s.GetState() {
	case apiv1.AutoscalePolicy_ENABLED:
		return "enabled"
	case apiv1.AutoscalePolicy_DISABLED:
		return "disabled"
	default:
		return "state_" + s.GetState().String()
	}
}

// listAutoscalePoliciesDirectRead returns a sentinel for the
// `ListAutoscalePolicies` RPC when the caller queries
// SHARED_POOL/CLUSTER scope. The autoscale_policies table is
// dormant for those scopes — see §Storage-point exclusivity —
// so the RPC surfaces a FailedPrecondition pointing the operator
// at the direct-read path (`GetEffectiveAutoscalePolicy` for the
// resolved view; `readRuntimeSpecs` for the raw spec list).
func listAutoscalePoliciesDirectReadRejected(
	subjectType entity.AutoscalePolicySubjectType,
	_ string,
	_ entity.ComponentType,
) (*entity.ListAutoscalePoliciesResponse, error) {
	if subjectType == apiv1.AutoscalePolicy_SHARED_POOL ||
		subjectType == apiv1.AutoscalePolicy_CLUSTER ||
		subjectType == apiv1.AutoscalePolicy_SUBJECT_TYPE_UNSPECIFIED {
		return nil, status.Errorf(codes.FailedPrecondition,
			"autoscale_policies is no longer a SHARED_POOL/CLUSTER storage point — see docs/design/tenant-aware-autoscale-policy-redesign.md §Storage-point exclusivity. Use GetEffectiveAutoscalePolicy for the resolved view.")
	}
	return nil, status.Errorf(codes.Unimplemented,
		"DEFAULT/TENANT scope policies are managed exclusively through the autoscale_policy_bundle_specs table; use ListAutoscalePolicy (the bundle RPC) instead")
}