package recorder

import (
	"bytes"
	"encoding/json"
	"os"
	"path/filepath"
	"reflect"
	"testing"
	"time"

	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"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/policy"
	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/replicas"
	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/tier"
)

func TestDecisionRecordPreservesPolicyInputThroughJSON(t *testing.T) {
	as := validAutoscaler()
	as.UID = types.UID("autoscaler-uid")
	as.Generation = 7
	as.Spec.Metrics = append(as.Spec.Metrics, v1alpha1.MetricSpec{
		Type: v1alpha1.MetricSourceTypeResource,
		Resource: &v1alpha1.ResourceMetricSource{
			Name: corev1.ResourceMemory,
			Target: v1alpha1.MetricTarget{
				Type:               v1alpha1.MetricTargetTypeUtilization,
				AverageUtilization: int32Ptr(50),
			},
		},
	})

	small := &decisionTestTier{
		name: "small",
		resources: []tier.Resource{
			{Name: string(corev1.ResourceCPU), Capacity: resource.MustParse("2")},
			{Name: string(corev1.ResourceMemory), Capacity: resource.MustParse("8Gi")},
		},
		cost: 1,
	}
	large := &decisionTestTier{
		name: "large",
		resources: []tier.Resource{
			{Name: string(corev1.ResourceCPU), Capacity: resource.MustParse("4")},
			{Name: string(corev1.ResourceMemory), Capacity: resource.MustParse("16Gi")},
		},
		cost: 3,
	}
	tiers := []tier.Tier{small, large}
	promQLs := []string{"cpu", "memory"}
	metrics := []decisionmetrics.Metrics{
		{Series: []decisionmetrics.TimeSeries{
			{
				Ref: &decisionmetrics.Reference{Name: "tidb-0"},
				Samples: []decisionmetrics.Sample{
					{Timestamp: time.Unix(1, 0).UTC(), Value: 1},
					{Timestamp: time.Unix(2, 0).UTC(), Value: 2},
				},
			},
			{
				Ref:     &decisionmetrics.Reference{Name: "tidb-1"},
				Samples: []decisionmetrics.Sample{{Timestamp: time.Unix(3, 0).UTC(), Value: 1}},
			},
		}},
		{Series: []decisionmetrics.TimeSeries{
			{
				Ref:     nil,
				Samples: []decisionmetrics.Sample{{Timestamp: time.Unix(4, 0).UTC(), Value: float64(4 * 1024 * 1024 * 1024)}},
			},
			{
				Ref:     &decisionmetrics.Reference{Name: "tidb-1"},
				Samples: []decisionmetrics.Sample{{Timestamp: time.Unix(5, 0).UTC(), Value: float64(4 * 1024 * 1024 * 1024)}},
			},
		}},
	}

	calculators := decisionTestCalculators(t, as.Spec.Metrics)
	p, err := policy.NewCostOptimized(tiers, calculators, as.Spec.Replicas)
	if err != nil {
		t.Fatal(err)
	}
	targetTier, targetReplicas, err := p.Choose(metrics)
	if err != nil {
		t.Fatal(err)
	}

	record := NewDecision(as)
	record.Policy = v1alpha1.AutoscalingPolicyCostOptimized
	record.SetMetrics(as.Spec.Metrics, promQLs, metrics)
	record.SetTiers(tiers)
	record.SetRecommendation(tiers, targetTier, targetReplicas)
	data, err := json.Marshal(record)
	if err != nil {
		t.Fatalf("marshal decision record: %v", err)
	}
	var decoded Decision
	if err := json.Unmarshal(data, &decoded); err != nil {
		t.Fatalf("unmarshal decision record: %v", err)
	}

	if decoded.Autoscaler.UID != "autoscaler-uid" || decoded.Autoscaler.Generation != 7 {
		t.Fatalf("autoscaler identity = %+v", decoded.Autoscaler)
	}
	if decoded.PlannerVersion == "" || decoded.Policy != v1alpha1.AutoscalingPolicyCostOptimized {
		t.Fatalf("planner/policy = %q/%q", decoded.PlannerVersion, decoded.Policy)
	}
	if decoded.Target.Namespace != "dp-x" || decoded.Target.Kind != "TiDBGroup" || decoded.Target.Name != "tidb" {
		t.Fatalf("target = %+v", decoded.Target)
	}
	if decoded.ReplicaBounds == nil || decoded.ReplicaBounds.Min != 1 || decoded.ReplicaBounds.Max != 5 {
		t.Fatalf("replica bounds = %+v", decoded.ReplicaBounds)
	}
	if got := []string{decoded.Metrics[0].Query, decoded.Metrics[1].Query}; !reflect.DeepEqual(got, promQLs) {
		t.Fatalf("metric query order = %v, want %v", got, promQLs)
	}
	if decoded.Metrics[1].ResourceName != "memory" || decoded.Metrics[1].AverageUtilization != 50 {
		t.Fatalf("memory metric config = %+v", decoded.Metrics[1])
	}
	if decoded.Metrics[0].Series[0].Pod == nil || *decoded.Metrics[0].Series[0].Pod != "tidb-0" {
		t.Fatalf("first series pod = %v, want tidb-0", decoded.Metrics[0].Series[0].Pod)
	}
	if decoded.Metrics[1].Series[0].Pod != nil {
		t.Fatalf("nil series reference became %v", *decoded.Metrics[1].Series[0].Pod)
	}
	if got := decoded.Metrics[0].Series[0].Samples; !reflect.DeepEqual(got, []Sample{
		{Timestamp: time.Unix(1, 0).UTC(), Value: 1},
		{Timestamp: time.Unix(2, 0).UTC(), Value: 2},
	}) {
		t.Fatalf("sample order = %+v", got)
	}
	if got := []string{decoded.Tiers[0].Name, decoded.Tiers[1].Name}; !reflect.DeepEqual(got, []string{"small", "large"}) {
		t.Fatalf("tier order = %v", got)
	}
	if decoded.Tiers[0].Ordinal != 0 || decoded.Tiers[0].Resources[1].Capacity != "8Gi" || decoded.Tiers[0].Cost != 1 {
		t.Fatalf("small tier = %+v", decoded.Tiers[0])
	}
	if decoded.Recommendation.TierOrdinal != 0 || decoded.Recommendation.TierName != "small" || decoded.Recommendation.Replicas != 3 {
		t.Fatalf("recommendation = %+v", decoded.Recommendation)
	}

	replayedTier, replayedReplicas := replayDecisionRecord(t, &decoded)
	if replayedTier.Name() != decoded.Recommendation.TierName || replayedReplicas != decoded.Recommendation.Replicas {
		t.Fatalf("replayed decision = %s/%d, want %s/%d", replayedTier.Name(), replayedReplicas,
			decoded.Recommendation.TierName, decoded.Recommendation.Replicas)
	}
}

func TestDecisionRecordUsesEmptyArraysForEmptyMetricResult(t *testing.T) {
	as := validAutoscaler()
	candidate := &decisionTestTier{
		name:      "small",
		resources: []tier.Resource{{Name: string(corev1.ResourceCPU), Capacity: resource.MustParse("2")}},
		cost:      1,
	}
	record := NewDecision(as)
	record.Policy = v1alpha1.AutoscalingPolicyCostOptimized
	record.SetMetrics(as.Spec.Metrics, []string{"cpu"}, []decisionmetrics.Metrics{{}})
	record.SetTiers([]tier.Tier{candidate})
	record.SetRecommendation([]tier.Tier{candidate}, candidate, 1)

	data, err := json.Marshal(record)
	if err != nil {
		t.Fatalf("marshal decision record: %v", err)
	}
	if !json.Valid(data) {
		t.Fatalf("invalid decision record JSON: %s", data)
	}
	if got := record.Metrics[0].Series; got == nil || len(got) != 0 {
		t.Fatalf("empty metric series = %#v, want non-nil empty slice", got)
	}
	var decoded Decision
	if err := json.Unmarshal(data, &decoded); err != nil {
		t.Fatalf("unmarshal decision record: %v", err)
	}
	replayedTier, replayedReplicas := replayDecisionRecord(t, &decoded)
	if replayedTier.Name() != "small" || replayedReplicas != 1 {
		t.Fatalf("replayed empty-metric decision = %s/%d, want small/1", replayedTier.Name(), replayedReplicas)
	}
}

type decisionTestTier struct {
	name      string
	resources []tier.Resource
	cost      int64
}

var _ tier.Tier = (*decisionTestTier)(nil)

func (t *decisionTestTier) Name() string {
	return t.name
}

func (t *decisionTestTier) Resources() []tier.Resource {
	return t.resources
}

func (t *decisionTestTier) Capacity(name string) (resource.Quantity, bool) {
	for i := range t.resources {
		if t.resources[i].Name == name {
			return t.resources[i].Capacity.DeepCopy(), true
		}
	}
	return resource.Quantity{}, false
}

func (t *decisionTestTier) Patch() client.Patch {
	return nil
}

func (t *decisionTestTier) Cost() int64 {
	return t.cost
}

func decisionTestCalculators(t *testing.T, specs []v1alpha1.MetricSpec) []replicas.Calculator {
	t.Helper()
	calculators := make([]replicas.Calculator, len(specs))
	for i := range specs {
		source := specs[i].Resource
		calculator, err := replicas.NewTargetUtilizationCalculator(string(source.Name), *source.Target.AverageUtilization)
		if err != nil {
			t.Fatalf("calculator for metric %d: %v", i, err)
		}
		calculators[i] = calculator
	}
	return calculators
}

func replayDecisionRecord(t *testing.T, record *Decision) (tier.Tier, int32) {
	t.Helper()

	tiers := make([]tier.Tier, len(record.Tiers))
	for i := range record.Tiers {
		resources := make([]tier.Resource, len(record.Tiers[i].Resources))
		for j := range record.Tiers[i].Resources {
			resources[j] = tier.Resource{
				Name:     record.Tiers[i].Resources[j].Name,
				Capacity: resource.MustParse(record.Tiers[i].Resources[j].Capacity),
			}
		}
		tiers[i] = &decisionTestTier{name: record.Tiers[i].Name, resources: resources, cost: record.Tiers[i].Cost}
	}

	metrics := make([]decisionmetrics.Metrics, len(record.Metrics))
	calculators := make([]replicas.Calculator, len(record.Metrics))
	for i := range record.Metrics {
		metric := &record.Metrics[i]
		series := make([]decisionmetrics.TimeSeries, len(metric.Series))
		for j := range metric.Series {
			var ref *decisionmetrics.Reference
			if metric.Series[j].Pod != nil {
				ref = &decisionmetrics.Reference{Name: *metric.Series[j].Pod}
			}
			samples := make([]decisionmetrics.Sample, len(metric.Series[j].Samples))
			for k := range metric.Series[j].Samples {
				samples[k] = decisionmetrics.Sample(metric.Series[j].Samples[k])
			}
			series[j] = decisionmetrics.TimeSeries{Ref: ref, Samples: samples}
		}
		metrics[i] = decisionmetrics.Metrics{Series: series}

		calculator, err := replicas.NewTargetUtilizationCalculator(metric.ResourceName, metric.AverageUtilization)
		if err != nil {
			t.Fatalf("rebuild calculator %d: %v", i, err)
		}
		calculators[i] = calculator
	}

	var bounds *v1alpha1.ReplicasSpec
	if record.ReplicaBounds != nil {
		bounds = &v1alpha1.ReplicasSpec{Min: record.ReplicaBounds.Min, Max: record.ReplicaBounds.Max}
	}
	p, err := policy.NewCostOptimized(tiers, calculators, bounds)
	if err != nil {
		t.Fatalf("rebuild policy: %v", err)
	}
	targetTier, targetReplicas, err := p.Choose(metrics)
	if err != nil {
		t.Fatalf("replay decision: %v", err)
	}
	return targetTier, targetReplicas
}

func TestJSONLRecorderAppendsOneLinePerRecord(t *testing.T) {
	dir := t.TempDir()
	recorder, err := NewJSONLRecorder(dir)
	if err != nil {
		t.Fatal(err)
	}
	first := NewDecision(validAutoscaler())
	second := NewDecision(validAutoscaler())
	if err := recorder.Record(first); err != nil {
		t.Fatal(err)
	}
	if err := recorder.Record(second); err != nil {
		t.Fatal(err)
	}

	records := readDecisionRecords(t, dir)
	if len(records) != 2 {
		t.Fatalf("recorded decisions = %d, want 2", len(records))
	}
	if records[0].DecisionID == "" || records[0].DecisionID == records[1].DecisionID {
		t.Fatalf("decision ids = %q/%q, want distinct non-empty", records[0].DecisionID, records[1].DecisionID)
	}
}

func validAutoscaler() *v1alpha1.Autoscaler {
	return &v1alpha1.Autoscaler{
		ObjectMeta: metav1.ObjectMeta{Name: "tidb", Namespace: "dp-x"},
		Spec: v1alpha1.AutoscalerSpec{
			TargetRef: v1alpha1.CrossVersionObjectReference{
				APIVersion: "core.pingcap.com/v1alpha1",
				Kind:       "TiDBGroup",
				Name:       "tidb",
			},
			Mode:     v1alpha1.AutoscalerModeAuto,
			Replicas: &v1alpha1.ReplicasSpec{Min: 1, Max: 5},
			Policy:   v1alpha1.AutoscalingPolicyCostOptimized,
			Metrics: []v1alpha1.MetricSpec{{
				Type: v1alpha1.MetricSourceTypeResource,
				Resource: &v1alpha1.ResourceMetricSource{
					Name: corev1.ResourceCPU,
					Target: v1alpha1.MetricTarget{
						Type:               v1alpha1.MetricTargetTypeUtilization,
						AverageUtilization: int32Ptr(50),
					},
				},
			}},
		},
	}
}

func int32Ptr(value int32) *int32 {
	return &value
}

func readDecisionRecords(t *testing.T, dir string) []Decision {
	t.Helper()
	files, err := filepath.Glob(filepath.Join(dir, "decisions-*.jsonl"))
	if err != nil {
		t.Fatal(err)
	}
	var records []Decision
	for _, file := range files {
		data, err := os.ReadFile(file)
		if err != nil {
			t.Fatal(err)
		}
		for _, line := range bytes.Split(bytes.TrimSpace(data), []byte("\n")) {
			if len(line) == 0 {
				continue
			}
			var record Decision
			if err := json.Unmarshal(line, &record); err != nil {
				t.Fatalf("decode decision record %q: %v", string(line), err)
			}
			records = append(records, record)
		}
	}
	return records
}