package recorder

import (
	"context"
	"encoding/json"
	"sort"
	"time"

	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/util/uuid"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/api/v1alpha1"
	decisionmetrics "github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/metrics"
	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/tier"
	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/version"
)

// Decision outcomes for one active evaluation attempt. The evaluation
// framework's decisions.jsonl stream reads the outcome from the record itself
// instead of inferring it from the absence of one.
const (
	OutcomeOK               = "ok"
	OutcomeWaitingMetrics   = "waiting_metrics"
	OutcomeReadError        = "read_error"
	OutcomeComputationError = "computation_error"
	OutcomeExecutionError   = "execution_error"
)

// Decision is the structured input, context and output snapshot for one
// active evaluation. It is persisted for every attempt — success, the
// metrics-incomplete wait, and read/computation failures — so the export
// pipeline can consume one schema for AUTO and DRY_RUN modes.
type Decision struct {
	DecisionID     string                     `json:"decision_id"`
	DecisionTime   time.Time                  `json:"decision_time"`
	PlannerVersion string                     `json:"planner_version"`
	PlannerCommit  string                     `json:"planner_commit"`
	Mode           v1alpha1.AutoscalerMode    `json:"mode"`
	Outcome        string                     `json:"outcome"`
	Error          string                     `json:"error,omitempty"`
	Autoscaler     Autoscaler                 `json:"autoscaler"`
	Target         Target                     `json:"target"`
	Policy         v1alpha1.AutoscalingPolicy `json:"policy"`
	ReplicaBounds  *ReplicaBounds             `json:"replica_bounds"`
	Metrics        []Metric                   `json:"metrics"`
	Tiers          []Tier                     `json:"tiers"`
	// ObservedPods is the sorted name set returned by ListPods for this
	// evaluation — the exact set the metrics-completeness check compares
	// against. It is recorded independently of the best-effort current
	// capacity read so a waiting decision can be re-diagnosed offline.
	ObservedPods   []string         `json:"observed_pods"`
	Current        *CurrentCapacity `json:"current,omitempty"`
	Recommendation *Recommendation  `json:"recommendation,omitempty"`
}

type Autoscaler struct {
	Namespace  string `json:"namespace"`
	Name       string `json:"name"`
	UID        string `json:"uid"`
	Generation int64  `json:"generation"`
}

type Target struct {
	APIVersion string `json:"api_version"`
	Kind       string `json:"kind"`
	Namespace  string `json:"namespace"`
	Name       string `json:"name"`
}

type ReplicaBounds struct {
	Min int32 `json:"min"`
	Max int32 `json:"max"`
}

type CurrentCapacity struct {
	// TierOrdinal is -1 when the live target matches none of the configured
	// tiers, for example after a manual customization.
	TierOrdinal int `json:"tier_ordinal"`
	// Replicas is the desired replica count from the live target's
	// spec.replicas, not an observed count.
	Replicas *int32 `json:"replicas,omitempty"`
	// Pods and ReadyPods describe the metrics target Pod set observed at
	// decision time.
	Pods      int `json:"pods"`
	ReadyPods int `json:"ready_pods"`
}

type Metric struct {
	Query              string                    `json:"query"`
	SourceType         v1alpha1.MetricSourceType `json:"source_type"`
	ResourceName       string                    `json:"resource_name"`
	TargetType         v1alpha1.MetricTargetType `json:"target_type"`
	AverageUtilization int32                     `json:"average_utilization"`
	Series             []Series                  `json:"series"`
}

type Series struct {
	Pod     *string  `json:"pod"`
	Samples []Sample `json:"samples"`
}

type Sample struct {
	Timestamp time.Time `json:"timestamp"`
	Value     float64   `json:"value"`
}

type Tier struct {
	Ordinal   int            `json:"ordinal"`
	Name      string         `json:"name"`
	Resources []TierResource `json:"resources"`
	Cost      int64          `json:"cost"`
}

type TierResource struct {
	Name     string `json:"name"`
	Capacity string `json:"capacity"`
}

type Recommendation struct {
	TierOrdinal int    `json:"tier_ordinal"`
	TierName    string `json:"tier_name"`
	Replicas    int32  `json:"replicas"`
}

// NewDecision captures the identity and configuration context that
// is known before any metric is read. Policy starts as the raw spec value and
// is replaced with the effective policy once it resolves. DecisionTime is the
// record's occurrence time: it stays set even when evaluation fails before
// the metrics read.
func NewDecision(as *v1alpha1.Autoscaler) *Decision {
	decision := &Decision{
		DecisionID:     string(uuid.NewUUID()),
		DecisionTime:   time.Now().UTC(),
		PlannerVersion: version.Version(),
		PlannerCommit:  version.Commit(),
		Mode:           as.Spec.Mode,
		Policy:         as.Spec.Policy,
		ObservedPods:   []string{},
		Autoscaler: Autoscaler{
			Namespace:  as.Namespace,
			Name:       as.Name,
			UID:        string(as.UID),
			Generation: as.Generation,
		},
		Target: Target{
			APIVersion: as.Spec.TargetRef.APIVersion,
			Kind:       as.Spec.TargetRef.Kind,
			Namespace:  as.Namespace,
			Name:       as.Spec.TargetRef.Name,
		},
	}
	if as.Spec.Replicas != nil {
		decision.ReplicaBounds = &ReplicaBounds{
			Min: as.Spec.Replicas.Min,
			Max: as.Spec.Replicas.Max,
		}
	}
	return decision
}

// Finish sets the final outcome and optional error before persistence.
func (d *Decision) Finish(outcome string, cause error) {
	d.Outcome = outcome
	if cause != nil {
		d.Error = cause.Error()
	}
}

// currentCapacity observes the live target capacity at decision time. It is
// best-effort: a missing or unreadable target yields nil instead of blocking
// the decision. Replicas comes from spec.replicas (desired, not observed);
// Pods/ReadyPods come from the metrics target Pod set listed for this
// decision.
func (d *Decision) SetCurrentCapacity(ctx context.Context, reader client.Reader, as *v1alpha1.Autoscaler, tiers []tier.Tier, pods []*corev1.Pod) {
	gv, err := schema.ParseGroupVersion(as.Spec.TargetRef.APIVersion)
	if err != nil {
		return
	}
	live := &unstructured.Unstructured{}
	live.SetGroupVersionKind(gv.WithKind(as.Spec.TargetRef.Kind))
	key := client.ObjectKey{Namespace: as.Namespace, Name: as.Spec.TargetRef.Name}
	if err := reader.Get(ctx, key, live); err != nil {
		return
	}
	current := &CurrentCapacity{
		TierOrdinal: -1,
		Pods:        len(pods),
		ReadyPods:   readyPods(pods),
	}
	if replicas, ok, _ := unstructured.NestedInt64(live.Object, "spec", "replicas"); ok {
		value := int32(replicas)
		current.Replicas = &value
	}
	labels, _, _ := unstructured.NestedStringMap(live.Object, "metadata", "labels")
	for i := range tiers {
		if tierPatchLabelsMatch(tiers[i], labels) {
			current.TierOrdinal = i
			break
		}
	}
	d.Current = current
}

func readyPods(pods []*corev1.Pod) int {
	ready := 0
	for _, pod := range pods {
		for _, condition := range pod.Status.Conditions {
			if condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue {
				ready++
				break
			}
		}
	}
	return ready
}

// podNames returns the sorted Pod name set used by the metrics-completeness
// check, so a recorded decision pins the exact comparison set.
func (d *Decision) SetObservedPods(pods []*corev1.Pod) {
	names := make([]string, 0, len(pods))
	for _, pod := range pods {
		names = append(names, pod.Name)
	}
	sort.Strings(names)
	d.ObservedPods = names
}

// tierPatchLabelsMatch reports whether every label the tier patch writes to
// the target's metadata is already present with the same value on the live
// target, treating that label set as the tier's identity. This depends on the
// template writing identifying labels: a template without metadata labels
// never matches (ordinal stays -1), and a target whose labels changed before
// rollout converged, or whose containers were edited without updating the
// labels, is attributed to the tier the labels claim, not the tier the
// resources actually run.
func tierPatchLabelsMatch(t tier.Tier, liveLabels map[string]string) bool {
	patch := t.Patch()
	if patch == nil {
		return false
	}
	data, err := patch.Data(&unstructured.Unstructured{})
	if err != nil {
		return false
	}
	doc := map[string]interface{}{}
	if err := json.Unmarshal(data, &doc); err != nil {
		return false
	}
	metadata, ok := doc["metadata"].(map[string]interface{})
	if !ok {
		return false
	}
	patchLabels, ok := metadata["labels"].(map[string]interface{})
	if !ok || len(patchLabels) == 0 {
		return false
	}
	for key, value := range patchLabels {
		text, ok := value.(string)
		if !ok || liveLabels[key] != text {
			return false
		}
	}
	return true
}

func (d *Decision) SetMetrics(
	specs []v1alpha1.MetricSpec,
	promQLs []string,
	metrics []decisionmetrics.Metrics,
) {
	// A misbehaving Provider may return fewer series than requested; never
	// index specs beyond what was actually returned.
	count := len(metrics)
	if len(specs) < count {
		count = len(specs)
	}
	records := make([]Metric, count)
	for i := 0; i < count; i++ {
		source := specs[i].Resource
		records[i] = Metric{
			Query:        promQLs[i],
			SourceType:   specs[i].Type,
			ResourceName: string(source.Name),
			TargetType:   source.Target.Type,
			// The webhook rejects a nil AverageUtilization, so the value is
			// safe to dereference for any Autoscaler that reached the cluster.
			AverageUtilization: *source.Target.AverageUtilization,
			Series:             seriesRecords(metrics[i].Series),
		}
	}
	d.Metrics = records
}

func seriesRecords(series []decisionmetrics.TimeSeries) []Series {
	records := make([]Series, len(series))
	for i := range series {
		var pod *string
		if series[i].Ref != nil {
			name := series[i].Ref.Name
			pod = &name
		}
		samples := make([]Sample, len(series[i].Samples))
		for j := range series[i].Samples {
			samples[j] = Sample(series[i].Samples[j])
		}
		records[i] = Series{Pod: pod, Samples: samples}
	}
	return records
}

func (d *Decision) SetTiers(tiers []tier.Tier) {
	records := make([]Tier, len(tiers))
	for i := range tiers {
		resources := tiers[i].Resources()
		resourceRecords := make([]TierResource, len(resources))
		for j := range resources {
			resourceRecords[j] = TierResource{
				Name:     resources[j].Name,
				Capacity: resources[j].Capacity.String(),
			}
		}
		records[i] = Tier{
			Ordinal:   i,
			Name:      tiers[i].Name(),
			Resources: resourceRecords,
			Cost:      tiers[i].Cost(),
		}
	}
	d.Tiers = records
}

func (d *Decision) SetRecommendation(tiers []tier.Tier, targetTier tier.Tier, replicas int32) {
	d.Recommendation = &Recommendation{
		TierOrdinal: tierOrdinal(tiers, targetTier.Name()),
		TierName:    targetTier.Name(),
		Replicas:    replicas,
	}
}

func tierOrdinal(tiers []tier.Tier, targetTierName string) int {
	for i := range tiers {
		if tiers[i].Name() == targetTierName {
			return i
		}
	}
	return -1
}