package store

import (
	"context"
	"errors"
	"fmt"
	"math"
	"sort"
	"strings"
	"time"

	"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
	"github.com/jackc/pgx/v5"
)

type ListModelCandidatesOptions struct {
	CacheAffinityKey    string
	CacheAffinityKeys   []string
	CacheAffinityPolicy map[string]any
	QuotaScopePrefix    string
}

func (s *Store) ListModelCandidates(ctx context.Context, model string, modelType string, user *auth.User, options ...ListModelCandidatesOptions) ([]RuntimeModelCandidate, error) {
	exactModel := strings.TrimSpace(model)
	listOptions := normalizeListModelCandidatesOptions(modelType, options...)
	rows, err := s.pool.Query(ctx, `
	SELECT p.id::text, p.platform_key, p.name, p.provider,
       COALESCE(NULLIF(p.config->>'specType', ''), NULLIF(cp.provider_type, ''), NULLIF(p.config->>'sourceSpecType', ''), p.provider) AS spec_type,
       COALESCE(p.base_url, ''),
       p.auth_type, p.credentials, p.config, p.default_pricing_mode,
       p.default_discount_factor::float8, COALESCE(p.pricing_rule_set_id::text, ''),
       p.retry_policy, p.rate_limit_policy, COALESCE(b.default_rate_limit_policy, '{}'::jsonb),
       COALESCE(p.dynamic_priority, p.priority) AS effective_priority,
       m.id::text, COALESCE(m.base_model_id::text, ''), COALESCE(b.canonical_model_key, ''),
       COALESCE(NULLIF(m.provider_model_name, ''), m.model_name), m.model_name, COALESCE(m.model_alias, ''),
       $2::text AS requested_model_type, m.display_name,
       EXISTS (
         SELECT 1
         FROM model_compatibility_aliases compatibility_alias
         JOIN base_model_catalog alias_base ON alias_base.id = compatibility_alias.base_model_id
         WHERE compatibility_alias.alias = $1::text
           AND compatibility_alias.model_type = $2::text
           AND compatibility_alias.active = true
           AND (compatibility_alias.expires_at IS NULL OR compatibility_alias.expires_at > now())
           AND (compatibility_alias.base_model_id = b.id OR alias_base.invocation_name = b.invocation_name)
       ) AS legacy_alias_used,
       CASE
         WHEN b.capabilities #> '{text_generate,supportedApiProtocols}' IS NOT NULL
         THEN jsonb_set(
           COALESCE(b.capabilities, '{}'::jsonb) || COALESCE(m.capabilities, '{}'::jsonb),
           '{text_generate,supportedApiProtocols}',
           b.capabilities #> '{text_generate,supportedApiProtocols}',
           true
         )
         ELSE COALESCE(b.capabilities, '{}'::jsonb) || COALESCE(m.capabilities, '{}'::jsonb)
       END AS effective_capabilities,
       m.capability_override,
       COALESCE(b.base_billing_config, '{}'::jsonb), m.billing_config, m.billing_config_override,
       m.pricing_mode, COALESCE(m.discount_factor, 0)::float8, COALESCE(m.pricing_rule_set_id::text, ''),
       COALESCE(b.pricing_rule_set_id::text, ''),
       m.permission_config, m.retry_policy, m.rate_limit_policy, m.rate_limit_policy_mode,
       COALESCE(m.runtime_policy_set_id::text, COALESCE(b.runtime_policy_set_id::text, '')),
       (m.runtime_policy_set_id IS NOT NULL),
       COALESCE(NULLIF(m.runtime_policy_override, '{}'::jsonb), b.runtime_policy_override, '{}'::jsonb),
       COALESCE(m.runtime_policy_override, '{}'::jsonb),
       COALESCE(rp.retry_policy, '{}'::jsonb), COALESCE(rp.rate_limit_policy, '{}'::jsonb),
       COALESCE(rp.auto_disable_policy, '{}'::jsonb), COALESCE(rp.degrade_policy, '{}'::jsonb),
       COALESCE(con.active, 0)::float8,
       COALESCE(queued.waiting, 0)::float8,
       COALESCE(rpm.used_value, 0)::float8, COALESCE(rpm.reserved_value, 0)::float8,
       COALESCE(tpm.used_value, 0)::float8, COALESCE(tpm.reserved_value, 0)::float8,
	       COALESCE(s.running_count, 0)::float8,
	       COALESCE(s.waiting_count, 0)::float8,
	       COALESCE(s.limiter_ratio, 0)::float8,
	       COALESCE(EXTRACT(EPOCH FROM s.last_assigned_at), 0)::float8,
	       COALESCE(ca.cache_affinity_key, '')::text,
	       COALESCE(ca.request_count, 0)::float8,
	       COALESCE(ca.input_tokens, 0)::float8,
	       COALESCE(ca.cached_input_tokens, 0)::float8,
	       COALESCE(ca.ema_hit_ratio, 0)::float8,
	       COALESCE(ca.last_hit_ratio, 0)::float8,
	       COALESCE(EXTRACT(EPOCH FROM ca.last_observed_at), 0)::float8
	FROM platform_models m
	JOIN integration_platforms p ON p.id = m.platform_id
	LEFT JOIN model_catalog_providers cp ON cp.provider_key = p.provider OR cp.provider_code = p.provider
	LEFT JOIN base_model_catalog b ON b.id = m.base_model_id
	LEFT JOIN model_runtime_policy_sets rp ON rp.id = COALESCE(m.runtime_policy_set_id, b.runtime_policy_set_id)
	LEFT JOIN runtime_client_states s
	  ON s.client_id = p.platform_key || ':' || $2::text || ':' || COALESCE(NULLIF(m.provider_model_name, ''), m.model_name)
	LEFT JOIN LATERAL (
	  SELECT ca.cache_affinity_key, ca.request_count, ca.input_tokens, ca.cached_input_tokens, ca.ema_hit_ratio, ca.last_hit_ratio, ca.last_observed_at
	  FROM unnest($3::text[]) WITH ORDINALITY AS affinity_keys(cache_affinity_key, affinity_rank)
	  JOIN gateway_cache_affinity_stats ca ON ca.cache_affinity_key = affinity_keys.cache_affinity_key
	  WHERE ca.client_id = p.platform_key || ':' || $2::text || ':' || COALESCE(NULLIF(m.provider_model_name, ''), m.model_name)
	    AND ($4::int <= 0 OR ca.last_observed_at >= now() - ($4::int * interval '1 second'))
	  ORDER BY affinity_keys.affinity_rank ASC, ca.cached_input_tokens DESC, ca.ema_hit_ratio DESC, ca.last_observed_at DESC
	  LIMIT 1
	) ca ON TRUE
LEFT JOIN (
	  SELECT scope_key, SUM(lease_value) AS active
	  FROM gateway_concurrency_leases
	  WHERE scope_type = 'platform_model'
	    AND released_at IS NULL
	    AND expires_at > statement_timestamp()
	  GROUP BY scope_key
	) con ON con.scope_key = $5::text || m.id::text
	LEFT JOIN (
	  SELECT queued_sources.platform_model_id, COUNT(DISTINCT queued_sources.task_id) AS waiting
	  FROM (
	    SELECT admission.task_id::text AS task_id,
	           admission.platform_model_id::text AS platform_model_id
	    FROM gateway_task_admissions admission
	    JOIN gateway_tasks admission_task ON admission_task.id = admission.task_id
	    WHERE admission.status = 'waiting'
	      AND admission_task.status = 'queued'
	      AND admission_task.next_run_at <= now()
	    UNION ALL
	    SELECT t.id::text AS task_id, qm.id::text AS platform_model_id
	    FROM gateway_tasks t
	    JOIN integration_platforms qp ON TRUE
	    JOIN platform_models qm ON qm.platform_id = qp.id
	    WHERE t.async_mode = true
	      AND t.status = 'queued'
	      AND NULLIF(t.model_type, '') IS NOT NULL
	      AND qm.model_type @> jsonb_build_array(t.model_type)
	      AND t.queue_key = qp.platform_key || ':' || t.model_type || ':' || COALESCE(NULLIF(qm.provider_model_name, ''), qm.model_name)
	      AND NOT EXISTS (SELECT 1 FROM gateway_task_attempts existing_attempt WHERE existing_attempt.task_id = t.id)
	    UNION ALL
	    SELECT latest.task_id::text AS task_id, latest.platform_model_id
	    FROM (
	      SELECT DISTINCT ON (a.task_id) a.task_id, a.platform_model_id::text AS platform_model_id
	      FROM gateway_tasks t
	      JOIN gateway_task_attempts a ON a.task_id = t.id
	      WHERE t.async_mode = true
	        AND t.status = 'queued'
	        AND a.platform_model_id IS NOT NULL
	      ORDER BY a.task_id, a.attempt_no DESC, a.started_at DESC
	    ) latest
	  ) queued_sources
	  GROUP BY queued_sources.platform_model_id
	) queued ON queued.platform_model_id = m.id::text
LEFT JOIN (
	  SELECT DISTINCT ON (scope_key) scope_key, used_value, reserved_value
	  FROM gateway_rate_limit_counters
	  WHERE scope_type = 'platform_model'
	    AND metric = 'rpm'
	    AND reset_at > now()
	  ORDER BY scope_key, window_start DESC
	) rpm ON rpm.scope_key = $5::text || m.id::text
LEFT JOIN (
	  SELECT scope_key, SUM(used_value) AS used_value, SUM(reserved_value) AS reserved_value
	  FROM gateway_rate_limit_counters
	  WHERE scope_type = 'platform_model'
	    AND metric LIKE 'tpm%'
	    AND reset_at > now()
	  GROUP BY scope_key
	) tpm ON tpm.scope_key = $5::text || m.id::text
WHERE p.status = 'enabled'
  AND p.deleted_at IS NULL
  AND m.enabled = true
  AND m.model_type @> jsonb_build_array($2::text)
  AND (p.cooldown_until IS NULL OR p.cooldown_until <= now())
  AND (m.cooldown_until IS NULL OR m.cooldown_until <= now())
	AND (
	  b.invocation_name = $1::text
	  OR (b.id IS NULL AND m.model_name = $1::text)
	  OR EXISTS (
	    SELECT 1
	    FROM model_compatibility_aliases compatibility_alias
	    JOIN base_model_catalog alias_base ON alias_base.id = compatibility_alias.base_model_id
	    WHERE compatibility_alias.alias = $1::text
	      AND compatibility_alias.model_type = $2::text
	      AND compatibility_alias.active = true
	      AND (compatibility_alias.expires_at IS NULL OR compatibility_alias.expires_at > now())
	      AND (compatibility_alias.base_model_id = b.id OR alias_base.invocation_name = b.invocation_name)
	  )
	)
	ORDER BY effective_priority ASC,
	         COALESCE(s.running_count, 0) ASC,
	         COALESCE(s.waiting_count, 0) ASC,
	         COALESCE(s.last_assigned_at, to_timestamp(0)) ASC,
	         m.created_at ASC`, exactModel, modelType, listOptions.CacheAffinityKeys, cacheAffinityStaleAfterSeconds(listOptions.CacheAffinityPolicy), listOptions.QuotaScopePrefix)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := make([]RuntimeModelCandidate, 0)
	for rows.Next() {
		var item RuntimeModelCandidate
		var credentials []byte
		var platformConfig []byte
		var platformRetryPolicy []byte
		var platformRateLimitPolicy []byte
		var baseRateLimitPolicy []byte
		var capabilities []byte
		var capabilityOverride []byte
		var baseBilling []byte
		var billing []byte
		var billingOverride []byte
		var permissionConfig []byte
		var modelRetryPolicy []byte
		var modelRateLimitPolicy []byte
		var runtimePolicyOverride []byte
		var rateLimitRuntimeOverride []byte
		var runtimeRetryPolicy []byte
		var runtimeRateLimitPolicy []byte
		var autoDisablePolicy []byte
		var degradePolicy []byte
		var concurrentActive float64
		var queuedWaiting float64
		var rpmUsed float64
		var rpmReserved float64
		var tpmUsed float64
		var tpmReserved float64
		var stateRunningCount float64
		var stateWaitingCount float64
		var stateLimiterRatio float64
		var lastAssignedUnix float64
		var cacheAffinityKey string
		var cacheRequestCount float64
		var cacheInputTokens float64
		var cacheCachedInputTokens float64
		var cacheEMAHitRatio float64
		var cacheLastHitRatio float64
		var cacheLastObservedUnix float64
		if err := rows.Scan(
			&item.PlatformID,
			&item.PlatformKey,
			&item.PlatformName,
			&item.Provider,
			&item.SpecType,
			&item.BaseURL,
			&item.AuthType,
			&credentials,
			&platformConfig,
			&item.DefaultPricingMode,
			&item.DefaultDiscountFactor,
			&item.PlatformPricingRuleSetID,
			&platformRetryPolicy,
			&platformRateLimitPolicy,
			&baseRateLimitPolicy,
			&item.PlatformPriority,
			&item.PlatformModelID,
			&item.BaseModelID,
			&item.CanonicalModelKey,
			&item.ProviderModelName,
			&item.ModelName,
			&item.ModelAlias,
			&item.ModelType,
			&item.DisplayName,
			&item.LegacyAliasUsed,
			&capabilities,
			&capabilityOverride,
			&baseBilling,
			&billing,
			&billingOverride,
			&item.PricingMode,
			&item.DiscountFactor,
			&item.ModelPricingRuleSetID,
			&item.BasePricingRuleSetID,
			&permissionConfig,
			&modelRetryPolicy,
			&modelRateLimitPolicy,
			&item.ModelRateLimitPolicyMode,
			&item.RuntimePolicySetID,
			&item.RuntimePolicyExplicit,
			&runtimePolicyOverride,
			&rateLimitRuntimeOverride,
			&runtimeRetryPolicy,
			&runtimeRateLimitPolicy,
			&autoDisablePolicy,
			&degradePolicy,
			&concurrentActive,
			&queuedWaiting,
			&rpmUsed,
			&rpmReserved,
			&tpmUsed,
			&tpmReserved,
			&stateRunningCount,
			&stateWaitingCount,
			&stateLimiterRatio,
			&lastAssignedUnix,
			&cacheAffinityKey,
			&cacheRequestCount,
			&cacheInputTokens,
			&cacheCachedInputTokens,
			&cacheEMAHitRatio,
			&cacheLastHitRatio,
			&cacheLastObservedUnix,
		); err != nil {
			return nil, err
		}
		item.Credentials = decodeObject(credentials)
		item.PlatformConfig = decodeObject(platformConfig)
		item.PlatformRetryPolicy = decodeObject(platformRetryPolicy)
		item.PlatformRateLimitPolicy = decodeObject(platformRateLimitPolicy)
		item.BaseRateLimitPolicy = decodeObject(baseRateLimitPolicy)
		item.Capabilities = decodeObject(capabilities)
		item.CapabilityOverride = decodeObject(capabilityOverride)
		item.BaseBillingConfig = decodeObject(baseBilling)
		item.BillingConfig = decodeObject(billing)
		item.BillingConfigOverride = decodeObject(billingOverride)
		item.PermissionConfig = decodeObject(permissionConfig)
		item.ModelRetryPolicy = decodeObject(modelRetryPolicy)
		item.ModelRateLimitPolicy = decodeObject(modelRateLimitPolicy)
		item.RuntimePolicyOverride = decodeObject(runtimePolicyOverride)
		item.RateLimitRuntimeOverride = decodeObject(rateLimitRuntimeOverride)
		item.RuntimeRetryPolicy = decodeObject(runtimeRetryPolicy)
		item.RuntimeRateLimitPolicy = decodeObject(runtimeRateLimitPolicy)
		item.AutoDisablePolicy = decodeObject(autoDisablePolicy)
		item.DegradePolicy = decodeObject(degradePolicy)
		upstreamModelName := firstNonEmpty(item.ProviderModelName, item.ModelName)
		item.ClientID = fmt.Sprintf("%s:%s:%s", item.PlatformKey, item.ModelType, upstreamModelName)
		item.QueueKey = item.ClientID
		item.RunningCount = stateRunningCount
		item.WaitingCount = maxFloat(queuedWaiting, stateWaitingCount)
		item.LastAssignedUnix = lastAssignedUnix
		applyRuntimeCandidateCacheAffinity(&item, listOptions, runtimeCandidateCacheAffinityInput{
			Key:               cacheAffinityKey,
			RequestCount:      int(cacheRequestCount),
			InputTokens:       int(cacheInputTokens),
			CachedInputTokens: int(cacheCachedInputTokens),
			EMAHitRatio:       cacheEMAHitRatio,
			LastHitRatio:      cacheLastHitRatio,
			LastObservedUnix:  cacheLastObservedUnix,
		})
		applyRuntimeCandidateLoad(&item, runtimeCandidateLoadInput{
			Policy: EffectiveRateLimitPolicy(EffectiveRateLimitPolicyInput{
				BasePolicy:            item.BaseRateLimitPolicy,
				PlatformPolicy:        item.PlatformRateLimitPolicy,
				RuntimePolicy:         item.RuntimeRateLimitPolicy,
				RuntimePolicyExplicit: item.RuntimePolicyExplicit,
				RuntimePolicyOverride: item.RateLimitRuntimeOverride,
				ModelPolicy:           item.ModelRateLimitPolicy,
				ModelPolicyMode:       item.ModelRateLimitPolicyMode,
			}),
			ConcurrentActive:  concurrentActive,
			QueuedWaiting:     queuedWaiting,
			RPMUsed:           rpmUsed,
			RPMReserved:       rpmReserved,
			TPMUsed:           tpmUsed,
			TPMReserved:       tpmReserved,
			StateRunningCount: stateRunningCount,
			StateWaitingCount: stateWaitingCount,
			StateLimiterRatio: stateLimiterRatio,
		})
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	if len(items) == 0 {
		if unavailableErr, err := s.modelCandidateCooldownError(ctx, model, modelType); err != nil {
			return nil, err
		} else if unavailableErr != nil {
			return nil, unavailableErr
		}
		return nil, ErrNoModelCandidate
	}
	items, err = s.filterCandidatesByAccessRules(ctx, user, items)
	if err != nil {
		return nil, err
	}
	if len(items) == 0 {
		return nil, ErrNoModelCandidate
	}
	s.recordLegacyAliasUsage(ctx, exactModel, items)
	sortRuntimeModelCandidates(items)
	return items, nil
}

func (s *Store) recordLegacyAliasUsage(ctx context.Context, alias string, items []RuntimeModelCandidate) {
	seen := map[string]bool{}
	for _, item := range items {
		if !item.LegacyAliasUsed || item.CanonicalModelKey == "" {
			continue
		}
		key := item.CanonicalModelKey + "\x00" + item.ModelType
		if seen[key] {
			continue
		}
		seen[key] = true
		_, _ = s.pool.Exec(ctx, `
INSERT INTO model_alias_usage_metrics (alias, canonical_model_key, model_type, hit_count)
VALUES ($1, $2, $3, 1)
ON CONFLICT (alias, canonical_model_key, model_type) DO UPDATE
SET hit_count = model_alias_usage_metrics.hit_count + 1,
    last_used_at = now()`, alias, item.CanonicalModelKey, item.ModelType)
	}
}

func (s *Store) GetRuntimeModelCandidateForVoiceCloneDeletion(ctx context.Context, platformModelID string, platformID string) (RuntimeModelCandidate, bool, error) {
	platformModelID = strings.TrimSpace(platformModelID)
	platformID = strings.TrimSpace(platformID)
	if platformModelID == "" && platformID == "" {
		return RuntimeModelCandidate{}, false, nil
	}
	var item RuntimeModelCandidate
	var credentials []byte
	var platformConfig []byte
	err := s.pool.QueryRow(ctx, `
SELECT p.id::text, p.platform_key, p.name, p.provider,
       COALESCE(NULLIF(p.config->>'specType', ''), NULLIF(cp.provider_type, ''), NULLIF(p.config->>'sourceSpecType', ''), p.provider) AS spec_type,
       COALESCE(p.base_url, ''), p.auth_type, p.credentials, p.config,
       COALESCE(m.id::text, ''), COALESCE(NULLIF(m.provider_model_name, ''), m.model_name, 'voice_clone'),
       COALESCE(m.model_name, 'voice_clone'), COALESCE(m.model_alias, '')
FROM integration_platforms p
LEFT JOIN platform_models m
  ON m.platform_id = p.id
  AND (
    (NULLIF($1, '')::uuid IS NOT NULL AND m.id = NULLIF($1, '')::uuid)
    OR (
      NULLIF($1, '') IS NULL
      AND m.model_type @> jsonb_build_array('voice_clone'::text)
    )
  )
LEFT JOIN model_catalog_providers cp ON cp.provider_key = p.provider OR cp.provider_code = p.provider
WHERE p.deleted_at IS NULL
  AND (
    (NULLIF($1, '')::uuid IS NOT NULL AND m.id = NULLIF($1, '')::uuid)
    OR (NULLIF($2, '')::uuid IS NOT NULL AND p.id = NULLIF($2, '')::uuid)
  )
ORDER BY CASE WHEN NULLIF($1, '')::uuid IS NOT NULL AND m.id = NULLIF($1, '')::uuid THEN 0 ELSE 1 END,
         m.created_at DESC
LIMIT 1`, platformModelID, platformID).Scan(
		&item.PlatformID,
		&item.PlatformKey,
		&item.PlatformName,
		&item.Provider,
		&item.SpecType,
		&item.BaseURL,
		&item.AuthType,
		&credentials,
		&platformConfig,
		&item.PlatformModelID,
		&item.ProviderModelName,
		&item.ModelName,
		&item.ModelAlias,
	)
	if err != nil {
		if IsNotFound(err) {
			return RuntimeModelCandidate{}, false, nil
		}
		return RuntimeModelCandidate{}, false, err
	}
	item.Credentials = decodeObject(credentials)
	item.PlatformConfig = decodeObject(platformConfig)
	item.ModelType = "voice_clone"
	item.ClientID = item.PlatformKey + ":voice_clone:" + firstNonEmpty(item.ProviderModelName, item.ModelName)
	item.QueueKey = item.ClientID
	return item, true, nil
}

type runtimeCandidateLoadInput struct {
	Policy            map[string]any
	ConcurrentActive  float64
	QueuedWaiting     float64
	RPMUsed           float64
	RPMReserved       float64
	TPMUsed           float64
	TPMReserved       float64
	StateRunningCount float64
	StateWaitingCount float64
	StateLimiterRatio float64
}

type runtimeCandidateCacheAffinityInput struct {
	Key               string
	RequestCount      int
	InputTokens       int
	CachedInputTokens int
	EMAHitRatio       float64
	LastHitRatio      float64
	LastObservedUnix  float64
}

func normalizeListModelCandidatesOptions(modelType string, options ...ListModelCandidatesOptions) ListModelCandidatesOptions {
	if len(options) == 0 {
		return ListModelCandidatesOptions{}
	}
	out := options[0]
	out.CacheAffinityKey = strings.TrimSpace(out.CacheAffinityKey)
	out.QuotaScopePrefix = strings.TrimSpace(out.QuotaScopePrefix)
	out.CacheAffinityKeys = normalizedCacheAffinityKeys(out.CacheAffinityKey, out.CacheAffinityKeys)
	if len(out.CacheAffinityKeys) > 0 {
		out.CacheAffinityKey = out.CacheAffinityKeys[0]
	}
	if len(out.CacheAffinityKeys) == 0 || !cacheAffinityPolicyEnabled(out.CacheAffinityPolicy, modelType) {
		out.CacheAffinityKey = ""
		out.CacheAffinityKeys = nil
	}
	return out
}

func normalizedCacheAffinityKeys(primary string, keys []string) []string {
	seen := map[string]bool{}
	out := make([]string, 0, len(keys)+1)
	for _, key := range append([]string{primary}, keys...) {
		key = strings.TrimSpace(key)
		if key == "" || seen[key] {
			continue
		}
		seen[key] = true
		out = append(out, key)
	}
	return out
}

func applyRuntimeCandidateCacheAffinity(candidate *RuntimeModelCandidate, options ListModelCandidatesOptions, input runtimeCandidateCacheAffinityInput) {
	key := strings.TrimSpace(input.Key)
	if key == "" {
		key = options.CacheAffinityKey
	}
	affinity := RuntimeCandidateCacheAffinity{
		Key:                key,
		RequestCount:       input.RequestCount,
		InputTokens:        input.InputTokens,
		CachedInputTokens:  input.CachedInputTokens,
		EMAHitRatio:        boundedRatio(input.EMAHitRatio),
		LastHitRatio:       boundedRatio(input.LastHitRatio),
		LastObservedUnix:   input.LastObservedUnix,
		AdjustedPriority:   float64(candidate.PlatformPriority),
		MatchedPrefixDepth: cacheAffinityMatchedPrefixDepth(key, options.CacheAffinityKeys),
	}
	minSamples := cacheAffinityMinSamples(options.CacheAffinityPolicy)
	hasObservedCachedHit := affinity.CachedInputTokens > 0 || affinity.LastHitRatio > 0
	hasSampledAffinity := input.RequestCount >= minSamples && affinity.EMAHitRatio > 0
	if key == "" || input.RequestCount <= 0 || (!hasObservedCachedHit && !hasSampledAffinity) {
		candidate.CacheAffinity = affinity
		return
	}
	affinity.Confidence = float64(input.RequestCount) / float64(minSamples)
	if affinity.Confidence > 1 {
		affinity.Confidence = 1
	}
	if hasObservedCachedHit && affinity.Confidence < 1 {
		affinity.Confidence = 1
	}
	affinity.Score = runtimeCandidateCacheAffinityScore(affinity)
	maxBoost := cacheAffinityMaxPriorityBoost(options.CacheAffinityPolicy)
	affinity.Boost = affinity.Score * maxBoost
	if affinity.Boost > maxBoost {
		affinity.Boost = maxBoost
	}
	if hasObservedCachedHit || affinity.Score > 0 {
		affinity.Applied = true
	}
	if affinity.Boost > 0 {
		affinity.AdjustedPriority = float64(candidate.PlatformPriority) - affinity.Boost
	}
	candidate.CacheAffinity = affinity
}

func cacheAffinityMatchedPrefixDepth(key string, lookup []string) int {
	prefix := ""
	switch {
	case strings.HasPrefix(key, "prompt_lcp_v2:"):
		prefix = "prompt_lcp_v2:"
	case strings.HasPrefix(key, "prompt_lcp:"):
		prefix = "prompt_lcp:"
	default:
		return 0
	}
	matchingKeys := make([]string, 0, len(lookup))
	for _, candidateKey := range lookup {
		if strings.HasPrefix(candidateKey, prefix) {
			matchingKeys = append(matchingKeys, candidateKey)
		}
	}
	for index, candidateKey := range matchingKeys {
		if candidateKey == key {
			return len(matchingKeys) - index + cacheAffinityMinimumPrefixDepth - 1
		}
	}
	return 0
}

const cacheAffinityMinimumPrefixDepth = 2

func cacheAffinityPolicyEnabled(policy map[string]any, modelType string) bool {
	if enabled, ok := policy["enabled"].(bool); ok && !enabled {
		return false
	}
	modelTypes := stringListFromAny(policy["modelTypes"])
	if len(modelTypes) == 0 {
		modelTypes = []string{"chat", "text_generate", "responses"}
	}
	modelType = strings.TrimSpace(modelType)
	for _, candidate := range modelTypes {
		if strings.EqualFold(strings.TrimSpace(candidate), modelType) {
			return true
		}
	}
	return false
}

func cacheAffinityMinSamples(policy map[string]any) int {
	value := intFromStorePolicy(policy, "minSamples", 3)
	if value <= 0 {
		return 1
	}
	return value
}

func cacheAffinityMinInputTokens(policy map[string]any) int {
	value := intFromStorePolicy(policy, "minInputTokens", 512)
	if value < 0 {
		return 0
	}
	return value
}

func cacheAffinityStaleAfterSeconds(policy map[string]any) int {
	return intFromStorePolicy(policy, "staleAfterSeconds", 86400)
}

func cacheAffinityMaxPriorityBoost(policy map[string]any) float64 {
	value := floatFromStorePolicy(policy, "maxPriorityBoost", 20)
	if value < 0 {
		return 0
	}
	return value
}

func cacheAffinityEMAAlpha(policy map[string]any) float64 {
	value := floatFromStorePolicy(policy, "emaAlpha", 0.35)
	if value <= 0 || value > 1 {
		return 0.35
	}
	return value
}

func runtimeCandidateCacheAffinityScore(affinity RuntimeCandidateCacheAffinity) float64 {
	hitRatio := affinity.EMAHitRatio
	if affinity.LastHitRatio > hitRatio {
		hitRatio = affinity.LastHitRatio
	}
	hitRatio = boundedRatio(hitRatio)
	confidence := boundedRatio(affinity.Confidence)
	if hitRatio <= 0 || confidence <= 0 {
		return 0
	}
	tokenFactor := 0.0
	if affinity.CachedInputTokens > 0 {
		tokenFactor = float64(affinity.CachedInputTokens) / 8192
		if tokenFactor > 1 {
			tokenFactor = 1
		}
	}
	return math.Round(hitRatio*confidence*(1+tokenFactor)*1_000_000) / 1_000_000
}

func intFromStorePolicy(policy map[string]any, key string, fallback int) int {
	if policy == nil {
		return fallback
	}
	switch value := policy[key].(type) {
	case int:
		return value
	case int64:
		return int(value)
	case float64:
		return int(value)
	default:
		return fallback
	}
}

func floatFromStorePolicy(policy map[string]any, key string, fallback float64) float64 {
	if policy == nil {
		return fallback
	}
	switch value := policy[key].(type) {
	case int:
		return float64(value)
	case int64:
		return float64(value)
	case float64:
		return value
	default:
		return fallback
	}
}

func boundedRatio(value float64) float64 {
	if value < 0 {
		return 0
	}
	if value > 1 {
		return 1
	}
	return value
}

func applyRuntimeCandidateLoad(candidate *RuntimeModelCandidate, input runtimeCandidateLoadInput) {
	rpmLimit := rateLimitForMetric(input.Policy, "rpm")
	tpmLimitValue := tpmLimit(input.Policy)
	concurrentLimit := rateLimitForMetric(input.Policy, "concurrent")
	rpmCurrent := input.RPMUsed + input.RPMReserved
	tpmCurrent := input.TPMUsed + input.TPMReserved
	activeCurrent := maxFloat(input.ConcurrentActive, input.StateRunningCount)
	waitingCurrent := maxFloat(input.QueuedWaiting, input.StateWaitingCount)
	concurrentCurrent := activeCurrent + waitingCurrent
	softRatio := unboundedLoadRatio(concurrentCurrent)
	concurrentRatio := softRatio
	if concurrentLimit > 0 {
		concurrentRatio = ratioIfLimited(concurrentCurrent, concurrentLimit)
	}
	metrics := RuntimeCandidateLoadMetrics{
		RPMCurrent:        rpmCurrent,
		RPMLimit:          rpmLimit,
		RPMRatio:          ratioIfLimited(rpmCurrent, rpmLimit),
		TPMCurrent:        tpmCurrent,
		TPMLimit:          tpmLimitValue,
		TPMRatio:          ratioIfLimited(tpmCurrent, tpmLimitValue),
		ConcurrentCurrent: concurrentCurrent,
		ConcurrentLimit:   concurrentLimit,
		ConcurrentRatio:   concurrentRatio,
		SoftCurrent:       concurrentCurrent,
		SoftRatio:         softRatio,
		QueuedCount:       waitingCurrent,
		StateRunningCount: input.StateRunningCount,
		StateWaitingCount: input.StateWaitingCount,
		StateLimiterRatio: input.StateLimiterRatio,
	}
	candidate.RunningCount = activeCurrent
	candidate.WaitingCount = waitingCurrent
	candidate.LoadMetrics = metrics
	candidate.LoadLimited = rpmLimit > 0 || tpmLimitValue > 0 || concurrentLimit > 0
	candidate.LoadRatio = maxFloat(metrics.RPMRatio, metrics.TPMRatio, metrics.ConcurrentRatio)
	candidate.FullReasons = runtimeCandidateFullReasons(*candidate)
}

func ratioIfLimited(current float64, limit float64) float64 {
	if limit <= 0 {
		return 0
	}
	return current / limit
}

func unboundedLoadRatio(current float64) float64 {
	if current <= 0 {
		return 0
	}
	return current / (current + 1)
}

func sortRuntimeModelCandidates(items []RuntimeModelCandidate) {
	hasFull := false
	hasNonFull := false
	for index := range items {
		items[index].LoadAvoided = false
		items[index].SelectionReason = "normal_rotation"
		items[index].FullReasons = runtimeCandidateFullReasons(items[index])
		if !items[index].CacheAffinity.Applied && items[index].CacheAffinity.AdjustedPriority == 0 {
			items[index].CacheAffinity.AdjustedPriority = float64(items[index].PlatformPriority)
		}
		if runtimeCandidateFull(items[index]) {
			hasFull = true
		} else {
			hasNonFull = true
		}
	}
	if hasFull && hasNonFull {
		for index := range items {
			items[index].LoadAvoided = runtimeCandidateFull(items[index])
			if items[index].LoadAvoided {
				items[index].SelectionReason = "full_avoided"
			}
		}
	} else if hasFull {
		for index := range items {
			items[index].SelectionReason = "all_full_fallback"
		}
	}
	sort.SliceStable(items, func(i, j int) bool {
		aFull := runtimeCandidateFull(items[i])
		bFull := runtimeCandidateFull(items[j])
		if aFull != bFull {
			return !aFull
		}
		if items[i].PlatformPriority != items[j].PlatformPriority {
			return items[i].PlatformPriority < items[j].PlatformPriority
		}
		if items[i].CacheAffinity.Applied != items[j].CacheAffinity.Applied {
			return items[i].CacheAffinity.Applied
		}
		if items[i].CacheAffinity.Applied && items[j].CacheAffinity.Applied {
			if items[i].CacheAffinity.Score != items[j].CacheAffinity.Score {
				return items[i].CacheAffinity.Score > items[j].CacheAffinity.Score
			}
			if items[i].CacheAffinity.EMAHitRatio != items[j].CacheAffinity.EMAHitRatio {
				return items[i].CacheAffinity.EMAHitRatio > items[j].CacheAffinity.EMAHitRatio
			}
			if items[i].CacheAffinity.LastHitRatio != items[j].CacheAffinity.LastHitRatio {
				return items[i].CacheAffinity.LastHitRatio > items[j].CacheAffinity.LastHitRatio
			}
			if items[i].CacheAffinity.CachedInputTokens != items[j].CacheAffinity.CachedInputTokens {
				return items[i].CacheAffinity.CachedInputTokens > items[j].CacheAffinity.CachedInputTokens
			}
		}
		if items[i].CacheAffinity.AdjustedPriority != items[j].CacheAffinity.AdjustedPriority {
			return items[i].CacheAffinity.AdjustedPriority < items[j].CacheAffinity.AdjustedPriority
		}
		if items[i].LoadRatio != items[j].LoadRatio {
			return items[i].LoadRatio < items[j].LoadRatio
		}
		if items[i].WaitingCount != items[j].WaitingCount {
			return items[i].WaitingCount < items[j].WaitingCount
		}
		if items[i].RunningCount != items[j].RunningCount {
			return items[i].RunningCount < items[j].RunningCount
		}
		if items[i].LastAssignedUnix != items[j].LastAssignedUnix {
			return items[i].LastAssignedUnix < items[j].LastAssignedUnix
		}
		return items[i].PlatformModelID < items[j].PlatformModelID
	})
	appliedCount := 0
	for index := range items {
		if items[index].CacheAffinity.Applied {
			appliedCount++
		}
	}
	for index := range items {
		items[index].CacheAffinity.CandidateCount = appliedCount
	}
	if len(items) == 0 || items[0].CacheAffinity.Applied {
		return
	}
	for index := 1; index < len(items); index++ {
		if !items[index].CacheAffinity.Applied {
			continue
		}
		switch {
		case runtimeCandidateFull(items[index]) && !runtimeCandidateFull(items[0]):
			items[0].CacheAffinity.OverrideReason = "capacity_tier_unavailable"
		case items[index].PlatformPriority != items[0].PlatformPriority:
			items[0].CacheAffinity.OverrideReason = "different_effective_priority_tier"
		}
		if items[0].CacheAffinity.OverrideReason != "" {
			break
		}
	}
}

func runtimeCandidateFull(candidate RuntimeModelCandidate) bool {
	return len(runtimeCandidateFullReasons(candidate)) > 0
}

func runtimeCandidateFullReasons(candidate RuntimeModelCandidate) []string {
	if len(candidate.FullReasons) > 0 {
		return append([]string(nil), candidate.FullReasons...)
	}
	metrics := candidate.LoadMetrics
	reasons := make([]string, 0, 4)
	if metrics.ConcurrentLimit > 0 && metrics.ConcurrentRatio >= 1 {
		reasons = append(reasons, "concurrent")
	}
	if metrics.RPMLimit > 0 && metrics.RPMRatio >= 1 {
		reasons = append(reasons, "rpm")
	}
	if metrics.TPMLimit > 0 && metrics.TPMRatio >= 1 {
		reasons = append(reasons, "tpm")
	}
	if metrics.QueuedCount > 0 {
		reasons = append(reasons, "waiting")
	}
	if len(reasons) == 0 && candidate.LoadLimited && candidate.LoadRatio >= 1 {
		reasons = append(reasons, "limited")
	}
	return reasons
}

func (s *Store) modelCandidateCooldownError(ctx context.Context, model string, modelType string) (error, error) {
	exactModel := strings.TrimSpace(model)
	var code string
	var recoveryAt time.Time
	var remainingSeconds float64
	err := s.pool.QueryRow(ctx, `
SELECT CASE
         WHEN COALESCE(m.cooldown_until, to_timestamp(0)) >= COALESCE(p.cooldown_until, to_timestamp(0))
         THEN 'model_cooling_down'
         ELSE 'platform_cooling_down'
       END,
       GREATEST(COALESCE(p.cooldown_until, to_timestamp(0)), COALESCE(m.cooldown_until, to_timestamp(0))) AS recovery_at,
       GREATEST(
         EXTRACT(EPOCH FROM GREATEST(COALESCE(p.cooldown_until, to_timestamp(0)), COALESCE(m.cooldown_until, to_timestamp(0))) - now()),
         0
       )::float8
FROM platform_models m
JOIN integration_platforms p ON p.id = m.platform_id
LEFT JOIN base_model_catalog b ON b.id = m.base_model_id
WHERE p.status = 'enabled'
  AND p.deleted_at IS NULL
  AND m.enabled = true
  AND m.model_type @> jsonb_build_array($2::text)
	AND (
	  b.invocation_name = $1::text
	  OR (b.id IS NULL AND m.model_name = $1::text)
	  OR EXISTS (
	    SELECT 1
	    FROM model_compatibility_aliases compatibility_alias
	    JOIN base_model_catalog alias_base ON alias_base.id = compatibility_alias.base_model_id
	    WHERE compatibility_alias.alias = $1::text
	      AND compatibility_alias.model_type = $2::text
	      AND compatibility_alias.active = true
	      AND (compatibility_alias.expires_at IS NULL OR compatibility_alias.expires_at > now())
	      AND (compatibility_alias.base_model_id = b.id OR alias_base.invocation_name = b.invocation_name)
		  )
		)
  AND (
    p.cooldown_until > now()
    OR m.cooldown_until > now()
  )
ORDER BY GREATEST(COALESCE(p.cooldown_until, to_timestamp(0)), COALESCE(m.cooldown_until, to_timestamp(0))) ASC,
	         p.priority ASC,
	         m.created_at ASC
LIMIT 1`, exactModel, modelType).Scan(&code, &recoveryAt, &remainingSeconds)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	retryAfterSeconds := int(math.Ceil(remainingSeconds))
	if retryAfterSeconds < 1 {
		retryAfterSeconds = 1
	}
	message := "请求的模型暂时不可用，请稍后重试"
	if code == "platform_cooling_down" {
		message = "可用平台暂时不可用，请稍后重试"
	}
	recoveryAt = recoveryAt.UTC()
	return &ModelCandidateUnavailableError{
		Code:       code,
		Message:    message,
		RetryAfter: time.Duration(retryAfterSeconds) * time.Second,
		RecoveryAt: recoveryAt,
		Details: map[string]any{
			"retryAfterSeconds": retryAfterSeconds,
			"recoveryAt":        recoveryAt.Format(time.RFC3339),
		},
	}, nil
}
