package executor

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"testing"

	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"

	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/merger"
	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/tier"
)

type fakeTier struct {
	name      string
	resources []tier.Resource
	patch     client.Patch
	cost      int64
}

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

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

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

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

func (t *fakeTier) Patch() client.Patch {
	return t.patch
}

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

type executorSchemaResolver struct{}

func (executorSchemaResolver) Resolve(context.Context, schema.GroupVersionKind) (*apiextensionsv1.JSONSchemaProps, error) {
	return &apiextensionsv1.JSONSchemaProps{Type: "object"}, nil
}

func TestKubernetesExecutorMergesLatestTargetWithOptimisticLock(t *testing.T) {
	gvk := schema.GroupVersionKind{Group: "example.com", Version: "v1", Kind: "Workload"}
	baseClient := newTargetClient(t, gvk, map[string]interface{}{
		"metadata": map[string]interface{}{
			"labels": map[string]interface{}{"existing": "true"},
		},
		"spec": map[string]interface{}{"replicas": int64(1)},
	})

	var patchData []byte
	writer := interceptor.NewClient(baseClient, interceptor.Funcs{
		Patch: func(
			ctx context.Context,
			_ client.WithWatch,
			obj client.Object,
			patch client.Patch,
			opts ...client.PatchOption,
		) error {
			if patch.Type() != types.MergePatchType {
				t.Fatalf("patch type = %q, want merge patch", patch.Type())
			}
			var err error
			patchData, err = patch.Data(obj)
			if err != nil {
				t.Fatalf("patch data: %v", err)
			}
			return baseClient.Patch(ctx, obj, patch, opts...)
		},
	})
	executor := NewKubernetesExecutor(baseClient, writer, newObjectMerger(t))
	targetTier := &fakeTier{
		name: "4c",
		patch: client.RawPatch(types.MergePatchType, []byte(
			`{"apiVersion":"wrong/v1","kind":"Wrong","metadata":{"name":"wrong","labels":{"tier":"4c"}}}`,
		)),
	}
	target := &Target{Namespace: "default", APIVersion: "example.com/v1", Kind: "Workload", Name: "target"}

	if err := executor.Execute(context.Background(), target, targetTier, 3); err != nil {
		t.Fatalf("Execute: %v", err)
	}

	got := getTarget(t, baseClient, gvk)
	labels := got.GetLabels()
	if labels["existing"] != "true" || labels["tier"] != "4c" {
		t.Fatalf("labels = %v, want existing and tier labels", labels)
	}
	replicas, ok, err := unstructured.NestedInt64(got.Object, "spec", "replicas")
	if err != nil || !ok || replicas != 3 {
		t.Fatalf("spec.replicas = %d, %t, error %v; want 3", replicas, ok, err)
	}
	if got.GetName() != "target" || got.GetAPIVersion() != "example.com/v1" || got.GetKind() != "Workload" {
		t.Fatalf("identity = %s %s/%s, want target Workload", got.GetAPIVersion(), got.GetKind(), got.GetName())
	}
	var encodedPatch map[string]interface{}
	if err := json.Unmarshal(patchData, &encodedPatch); err != nil {
		t.Fatalf("decode patch: %v", err)
	}
	metadata := encodedPatch["metadata"].(map[string]interface{})
	if metadata["resourceVersion"] == nil || metadata["resourceVersion"] == "" {
		t.Fatalf("patch metadata = %v, want optimistic-lock resourceVersion", metadata)
	}
}

func TestKubernetesExecutorRetriesConflictFromLatestObject(t *testing.T) {
	gvk := schema.GroupVersionKind{Group: "example.com", Version: "v1", Kind: "Workload"}
	baseClient := newTargetClient(t, gvk, map[string]interface{}{
		"metadata": map[string]interface{}{"labels": map[string]interface{}{}},
		"spec":     map[string]interface{}{"replicas": int64(1)},
	})
	getCalls := 0
	reader := interceptor.NewClient(baseClient, interceptor.Funcs{
		Get: func(
			ctx context.Context,
			_ client.WithWatch,
			key client.ObjectKey,
			obj client.Object,
			opts ...client.GetOption,
		) error {
			getCalls++
			return baseClient.Get(ctx, key, obj, opts...)
		},
	})
	patchCalls := 0
	writer := interceptor.NewClient(baseClient, interceptor.Funcs{
		Patch: func(
			ctx context.Context,
			_ client.WithWatch,
			obj client.Object,
			patch client.Patch,
			opts ...client.PatchOption,
		) error {
			patchCalls++
			if patchCalls == 1 {
				return apierrors.NewConflict(
					schema.GroupResource{Group: "example.com", Resource: "workloads"},
					"target",
					errors.New("injected conflict"),
				)
			}
			return baseClient.Patch(ctx, obj, patch, opts...)
		},
	})
	executor := NewKubernetesExecutor(reader, writer, newObjectMerger(t))

	err := executor.Execute(
		context.Background(),
		&Target{Namespace: "default", APIVersion: "example.com/v1", Kind: "Workload", Name: "target"},
		&fakeTier{name: "tier", patch: client.RawPatch(types.MergePatchType, []byte(`{}`))},
		2,
	)
	if err != nil {
		t.Fatalf("Execute: %v", err)
	}
	if getCalls != 2 || patchCalls != 2 {
		t.Fatalf("calls = get %d, patch %d; want 2, 2", getCalls, patchCalls)
	}
}

func TestKubernetesExecutorRejectsInvalidInput(t *testing.T) {
	baseClient := fake.NewClientBuilder().Build()
	executor := NewKubernetesExecutor(baseClient, baseClient, newObjectMerger(t))
	err := executor.execute(context.Background(), &Target{}, &fakeTier{
		name:  "4c",
		patch: client.RawPatch(types.MergePatchType, []byte(`{}`)),
	}, 1)
	if err == nil || !strings.Contains(err.Error(), "target") {
		t.Fatalf("execute() error = %v, want target validation error", err)
	}
}

func TestKubernetesExecutorReturnsPatchError(t *testing.T) {
	gvk := schema.GroupVersionKind{Group: "example.com", Version: "v1", Kind: "Workload"}
	baseClient := newTargetClient(t, gvk, map[string]interface{}{
		"spec": map[string]interface{}{"replicas": int64(1)},
	})
	writer := interceptor.NewClient(baseClient, interceptor.Funcs{
		Patch: func(context.Context, client.WithWatch, client.Object, client.Patch, ...client.PatchOption) error {
			return errors.New("injected patch failure")
		},
	})
	err := NewKubernetesExecutor(baseClient, writer, newObjectMerger(t)).Execute(
		context.Background(),
		&Target{Namespace: "default", APIVersion: "example.com/v1", Kind: "Workload", Name: "target"},
		&fakeTier{name: "tier", patch: client.RawPatch(types.MergePatchType, []byte(`{}`))},
		1,
	)
	if err == nil || !strings.Contains(err.Error(), "injected patch failure") {
		t.Fatalf("Execute() error = %v, want patch failure", err)
	}
}

func newObjectMerger(t *testing.T) merger.Interface {
	t.Helper()
	m, err := merger.New(executorSchemaResolver{})
	if err != nil {
		t.Fatalf("merger.New: %v", err)
	}
	return m
}

func newTargetClient(t *testing.T, gvk schema.GroupVersionKind, object map[string]interface{}) client.WithWatch {
	t.Helper()
	scheme := runtime.NewScheme()
	scheme.AddKnownTypeWithName(gvk, &unstructured.Unstructured{})
	scheme.AddKnownTypeWithName(gvk.GroupVersion().WithKind(gvk.Kind+"List"), &unstructured.UnstructuredList{})
	target := &unstructured.Unstructured{Object: object}
	target.SetGroupVersionKind(gvk)
	target.SetNamespace("default")
	target.SetName("target")
	target.SetResourceVersion("1")
	return fake.NewClientBuilder().WithScheme(scheme).WithObjects(target).Build()
}

func getTarget(t *testing.T, c client.Client, gvk schema.GroupVersionKind) *unstructured.Unstructured {
	t.Helper()
	got := &unstructured.Unstructured{}
	got.SetGroupVersionKind(gvk)
	if err := c.Get(context.Background(), client.ObjectKey{Namespace: "default", Name: "target"}, got); err != nil {
		t.Fatalf("Get target: %v", err)
	}
	return got
}