package controller

import (
	"context"
	"fmt"
	"time"

	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/runtime"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/api/v1alpha1"
	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/executor"
	decisionmetrics "github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/metrics"
	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/policy"
	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/recorder"
	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/replicas"
	metricstarget "github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/target"
	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/tier"
)

const (
	DefaultSyncPeriod   = 30 * time.Second
	dryRunDecisionEvent = "capacity_planner_dry_run_decision"
	autoDecisionEvent   = "capacity_planner_auto_decision"
)

// Reconciler is the control-loop orchestrator for Autoscaler objects.
type Reconciler struct {
	client.Client
	Scheme          *runtime.Scheme
	TierTemplates   *v1alpha1.TierTemplates
	MetricsProvider decisionmetrics.Provider
	Executor        executor.Executor
	// Recorder persists AUTO and DRY_RUN decision records when configured.
	Recorder   recorder.Recorder
	SyncPeriod time.Duration
}

// Reconcile implements ctrl.Reconciler.
func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	var as v1alpha1.Autoscaler
	if err := r.Get(ctx, req.NamespacedName, &as); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}
	if !as.DeletionTimestamp.IsZero() || as.Spec.Mode == v1alpha1.AutoscalerModeDisabled {
		return ctrl.Result{}, nil
	}
	if as.Spec.Mode != v1alpha1.AutoscalerModeAuto && as.Spec.Mode != v1alpha1.AutoscalerModeDryRun {
		return ctrl.Result{}, fmt.Errorf("controller: mode %q is unsupported", as.Spec.Mode)
	}
	return r.reconcileActive(ctx, &as)
}

func (r *Reconciler) reconcileActive(ctx context.Context, as *v1alpha1.Autoscaler) (ctrl.Result, error) {
	if r.MetricsProvider == nil {
		return ctrl.Result{}, fmt.Errorf("controller: metrics provider is required")
	}
	if r.Executor == nil {
		return ctrl.Result{}, fmt.Errorf("controller: executor is required")
	}
	// Decision records are the collection surface described by the evaluation
	// framework; both AUTO and DRY_RUN produce them, distinguished by mode.
	decision := recorder.NewDecision(as)
	fail := func(outcome string, cause error) (ctrl.Result, error) {
		r.recordDecision(ctx, decision, "scaling decision", outcome, cause)
		return ctrl.Result{}, cause
	}

	effectivePolicy, err := resolvePolicy(as.Spec.Policy)
	if err != nil {
		return fail(recorder.OutcomeComputationError, err)
	}
	decision.Policy = effectivePolicy

	tiers, err := tier.New(r.TierTemplates, &as.Spec.Tiers)
	if err != nil {
		return fail(recorder.OutcomeComputationError, fmt.Errorf("controller: materialize tiers: %w", err))
	}
	decision.SetTiers(tiers)
	promQLs, calculators, err := calculatorsForMetrics(as.Spec.Metrics)
	if err != nil {
		return fail(recorder.OutcomeComputationError, err)
	}
	target, err := metricstarget.NewPodTargetForRef(as.Namespace, as.Spec.TargetRef)
	if err != nil {
		return fail(recorder.OutcomeComputationError, err)
	}

	currentMetrics, err := r.MetricsProvider.Query(ctx, target, promQLs...)
	if err != nil {
		return fail(recorder.OutcomeReadError, fmt.Errorf("controller: query metrics: %w", err))
	}
	decision.SetMetrics(as.Spec.Metrics, promQLs, currentMetrics)
	targetPods, err := metricstarget.ListPods(ctx, r.Client, target)
	if err != nil {
		return fail(recorder.OutcomeReadError, fmt.Errorf("controller: list metrics target Pods: %w", err))
	}
	decision.SetObservedPods(targetPods)
	decision.SetCurrentCapacity(ctx, r.Client, as, tiers, targetPods)
	// Missing Pod samples are unknown demand, not zero demand.
	if !metricsComplete(currentMetrics, len(promQLs), targetPods) {
		extras := []interface{}{
			"autoscaler_namespace", as.Namespace,
			"autoscaler_name", as.Name,
			"expected_pods", len(targetPods),
			"sample_pods", samplePodsByMetric(promQLs, currentMetrics),
		}
		r.recordDecision(ctx, decision, "waiting for decision metrics", recorder.OutcomeWaitingMetrics, nil, extras...)
		return ctrl.Result{RequeueAfter: r.syncPeriod()}, nil
	}
	p, err := policy.NewCostOptimized(tiers, calculators, as.Spec.Replicas)
	if err != nil {
		return fail(recorder.OutcomeComputationError, err)
	}
	targetTier, targetReplicas, err := p.Choose(currentMetrics)
	if err != nil {
		return fail(recorder.OutcomeComputationError, err)
	}
	decision.SetRecommendation(tiers, targetTier, targetReplicas)
	summary := []interface{}{
		"mode", as.Spec.Mode,
		"autoscaler_namespace", as.Namespace,
		"autoscaler_name", as.Name,
		"autoscaler_generation", as.Generation,
		"target_api_version", as.Spec.TargetRef.APIVersion,
		"target_kind", as.Spec.TargetRef.Kind,
		"target_namespace", as.Namespace,
		"target_name", as.Spec.TargetRef.Name,
		"policy", effectivePolicy,
		"sample_pods", samplePodsByMetric(promQLs, currentMetrics),
		"recommended_tier", targetTier.Name(),
		"recommended_replicas", targetReplicas,
	}
	if as.Spec.Mode == v1alpha1.AutoscalerModeDryRun {
		r.recordDecision(ctx, decision, "scaling decision", recorder.OutcomeOK, nil, summary...)
		return ctrl.Result{RequeueAfter: r.syncPeriod()}, nil
	}

	executeTarget := &executor.Target{
		Namespace:  as.Namespace,
		APIVersion: as.Spec.TargetRef.APIVersion,
		Kind:       as.Spec.TargetRef.Kind,
		Name:       as.Spec.TargetRef.Name,
	}
	if err := r.Executor.Execute(ctx, executeTarget, targetTier, targetReplicas); err != nil {
		executionErr := fmt.Errorf("controller: execute scaling decision: %w", err)
		r.recordDecision(ctx, decision, "scaling decision", recorder.OutcomeExecutionError, executionErr, summary...)
		return ctrl.Result{}, executionErr
	}
	r.recordDecision(ctx, decision, "scaling decision", recorder.OutcomeOK, nil, summary...)
	return ctrl.Result{RequeueAfter: r.syncPeriod()}, nil
}

// recordDecision persists the finalized record and emits its compact summary.
func (r *Reconciler) recordDecision(ctx context.Context, decision *recorder.Decision, msg, outcome string, cause error, extras ...interface{}) {
	decision.Finish(outcome, cause)
	logger := log.FromContext(ctx)
	if r.Recorder != nil {
		if err := r.Recorder.Record(decision); err != nil {
			logger.Error(err, "unable to write decision record", "decision_id", decision.DecisionID)
		}
	}
	event := dryRunDecisionEvent
	if decision.Mode == v1alpha1.AutoscalerModeAuto {
		event = autoDecisionEvent
	}
	kv := make([]interface{}, 0, len(extras)+6)
	kv = append(kv,
		"event", event,
		"decision_id", decision.DecisionID,
		"outcome", decision.Outcome,
	)
	kv = append(kv, extras...)
	logger.Info(msg, kv...)
}

func metricsComplete(current []decisionmetrics.Metrics, requiredMetrics int, pods []*corev1.Pod) bool {
	if len(current) != requiredMetrics || len(pods) == 0 {
		return false
	}

	expectedPods := make(map[string]struct{}, len(pods))
	for _, pod := range pods {
		expectedPods[pod.Name] = struct{}{}
	}

	for i := range current {
		observedPods := make(map[string]struct{}, len(current[i].Series))
		for j := range current[i].Series {
			series := &current[i].Series[j]
			if series.Ref == nil || len(series.Samples) == 0 {
				continue
			}
			if _, expected := expectedPods[series.Ref.Name]; !expected {
				return false
			}
			observedPods[series.Ref.Name] = struct{}{}
		}
		if len(observedPods) != len(expectedPods) {
			return false
		}
	}
	return true
}

func resolvePolicy(policy v1alpha1.AutoscalingPolicy) (v1alpha1.AutoscalingPolicy, error) {
	switch policy {
	case "", v1alpha1.AutoscalingPolicyCostOptimized:
		return v1alpha1.AutoscalingPolicyCostOptimized, nil
	default:
		return "", fmt.Errorf("controller: policy %q is unsupported", policy)
	}
}

func samplePodsByMetric(promQLs []string, ms []decisionmetrics.Metrics) map[string]int {
	counts := make(map[string]int, len(ms))
	for i := range ms {
		pods := make(map[string]struct{})
		for j := range ms[i].Series {
			if ref := ms[i].Series[j].Ref; ref != nil {
				pods[ref.Name] = struct{}{}
			}
		}
		if i < len(promQLs) {
			counts[promQLs[i]] = len(pods)
		}
	}
	return counts
}

func calculatorsForMetrics(specs []v1alpha1.MetricSpec) ([]string, []replicas.Calculator, error) {
	if len(specs) == 0 {
		return nil, nil, fmt.Errorf("controller: metrics are required")
	}

	promQLs := make([]string, 0, len(specs))
	calculators := make([]replicas.Calculator, 0, len(specs))
	seen := make(map[string]struct{}, len(specs))
	for i := range specs {
		name, calculator, err := calculatorForMetric(&specs[i])
		if err != nil {
			return nil, nil, fmt.Errorf("controller: metrics[%d]: %w", i, err)
		}
		if _, exists := seen[name]; exists {
			return nil, nil, fmt.Errorf("controller: metrics[%d]: duplicate metric %q", i, name)
		}
		seen[name] = struct{}{}
		promQLs = append(promQLs, name)
		calculators = append(calculators, calculator)
	}
	return promQLs, calculators, nil
}

func calculatorForMetric(spec *v1alpha1.MetricSpec) (string, replicas.Calculator, error) {
	switch spec.Type {
	case v1alpha1.MetricSourceTypeResource:
		return calculatorForResourceMetric(spec.Resource)
	default:
		return "", nil, fmt.Errorf("type %q is unsupported", spec.Type)
	}
}

func calculatorForResourceMetric(source *v1alpha1.ResourceMetricSource) (string, replicas.Calculator, error) {
	if source == nil {
		return "", nil, fmt.Errorf("resource is required")
	}
	if source.Name == "" {
		return "", nil, fmt.Errorf("resource.name is required")
	}
	target := &source.Target
	if target.Type != v1alpha1.MetricTargetTypeUtilization {
		return "", nil, fmt.Errorf("target.type %q is unsupported", target.Type)
	}
	if target.AverageUtilization == nil {
		return "", nil, fmt.Errorf("target.averageUtilization is required")
	}
	if *target.AverageUtilization > 100 {
		return "", nil, fmt.Errorf("target.averageUtilization must be between 1 and 100")
	}
	name := string(source.Name)
	calculator, err := replicas.NewTargetUtilizationCalculator(name, *target.AverageUtilization)
	if err != nil {
		return "", nil, err
	}
	return name, calculator, nil
}

func (r *Reconciler) syncPeriod() time.Duration {
	if r.SyncPeriod > 0 {
		return r.SyncPeriod
	}
	return DefaultSyncPeriod
}

// SetupWithManager registers the controller with a single worker and the
// Autoscaler as the only primary watch.
func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&v1alpha1.Autoscaler{}).
		WithOptions(controller.Options{MaxConcurrentReconciles: 1}).
		Complete(r)
}