package k8s

import (
	"context"
	_ "embed"
	"fmt"

	"github.com/imdario/mergo"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"

	"github.com/pingcap/test-infra/sdk/resource"
)

var (
	//go:embed assets/pod-failure.yaml
	podFailureSpec []byte
	//go:embed assets/network-partition.yaml
	networkPartitionSpec []byte
	//go:embed assets/network-delay.yaml
	networkDelaySpec []byte
	//go:embed assets/network-loss.yaml
	networkLossSpec []byte
	//go:embed assets/io-delay.yaml
	ioDelaySpec []byte
	//go:embed assets/time-chaos.yaml
	timeChaosSpec []byte
	//go:embed assets/stress-chaos.yaml
	stressChaosSpec []byte

	defaultPodFailureSpec       map[string]interface{}
	defaultNetworkPartitionSpec map[string]interface{}
	defaultNetworkDelaySpec     map[string]interface{}
	defaultNetworkLossSpec      map[string]interface{}
	defaultIODelaySpec          map[string]interface{}
	defaultTimeChaosSpec        map[string]interface{}
	defaultStressChaosSpec      map[string]interface{}

	podChaosGVR     = schema.GroupVersionResource{Group: "chaos-mesh.org", Version: "v1alpha1", Resource: "podchaos"}
	networkChaosGVR = schema.GroupVersionResource{Group: "chaos-mesh.org", Version: "v1alpha1", Resource: "networkchaos"}
	ioChaosGVR      = schema.GroupVersionResource{Group: "chaos-mesh.org", Version: "v1alpha1", Resource: "iochaos"}
	timeChaosGVR    = schema.GroupVersionResource{Group: "chaos-mesh.org", Version: "v1alpha1", Resource: "timechaos"}
	stressChaosGVR  = schema.GroupVersionResource{Group: "chaos-mesh.org", Version: "v1alpha1", Resource: "stresschaos"}
)

func init() {
	mustUnmarshalYaml(podFailureSpec, &defaultPodFailureSpec)
	mustUnmarshalYaml(networkPartitionSpec, &defaultNetworkPartitionSpec)
	mustUnmarshalYaml(networkDelaySpec, &defaultNetworkDelaySpec)
	mustUnmarshalYaml(networkLossSpec, &defaultNetworkLossSpec)
	mustUnmarshalYaml(ioDelaySpec, &defaultIODelaySpec)
	mustUnmarshalYaml(timeChaosSpec, &defaultTimeChaosSpec)
	mustUnmarshalYaml(stressChaosSpec, &defaultStressChaosSpec)
}

type ChaosClient struct {
	KubeClientSet
}

type ChaosIdentifier struct {
	Namespace string
	Name      string
	Spec      resource.ExperimentSpec
}

func (c *ChaosIdentifier) HumanReadableName() string {
	return fmt.Sprintf("ns=%s,kind=%s,name=%s,spec=%#v", c.Namespace, c.Spec.Kind(), c.Name, c)
}

func (c *ChaosClient) Create(opts resource.InjectChaosOptions) (resource.ChaosIdentifier, error) {
	switch opts.Kind() {
	case resource.ExperimentFailureKind:
		return c.createPodFailureObject(opts, opts.ExperimentSpec.(*resource.FailureExperimentSpec))
	case resource.ExperimentNetworkDelayKind:
		return c.createNetworkDelayObject(opts, opts.ExperimentSpec.(*resource.NetworkLatencySpec))
	case resource.ExperimentNetworkPartitionKind:
		return c.createNetworkPartitionObject(opts, opts.ExperimentSpec.(*resource.NetworkPartitionSpec))
	case resource.ExperimentNetworkLossKind:
		return c.createNetworkLossObject(opts, opts.ExperimentSpec.(*resource.NetworkLossSpec))
	case resource.ExperimentIODelayKind:
		return c.createIODelayObject(opts, opts.ExperimentSpec.(*resource.IODelaySpec))
	case resource.ExperimentTimeChaosKind:
		return c.createTimeChaosObject(opts, opts.ExperimentSpec.(*resource.TimeChaosSpec))
	case resource.ExperimentStressChaosKind:
		return c.createStressChaosObject(opts, opts.ExperimentSpec.(*resource.StressChaosSpec))
	}
	return nil, fmt.Errorf("unknown chaos kind: %s", opts.Kind())
}

func (c *ChaosClient) Delete(id resource.ChaosIdentifier) error {
	chaosId := id.(*ChaosIdentifier)
	switch chaosId.Spec.Kind() {
	case resource.ExperimentFailureKind:
		return c.DynamicClient.Resource(podChaosGVR).Namespace(c.Namespace).Delete(context.TODO(), chaosId.Name, metav1.DeleteOptions{})
	case resource.ExperimentNetworkDelayKind:
		return c.DynamicClient.Resource(networkChaosGVR).Namespace(c.Namespace).Delete(context.TODO(), chaosId.Name, metav1.DeleteOptions{})
	case resource.ExperimentNetworkPartitionKind:
		return c.DynamicClient.Resource(networkChaosGVR).Namespace(c.Namespace).Delete(context.TODO(), chaosId.Name, metav1.DeleteOptions{})
	case resource.ExperimentNetworkLossKind:
		return c.DynamicClient.Resource(networkChaosGVR).Namespace(c.Namespace).Delete(context.TODO(), chaosId.Name, metav1.DeleteOptions{})
	case resource.ExperimentIODelayKind:
		return c.DynamicClient.Resource(ioChaosGVR).Namespace(c.Namespace).Delete(context.TODO(), chaosId.Name, metav1.DeleteOptions{})
	case resource.ExperimentTimeChaosKind:
		return c.DynamicClient.Resource(timeChaosGVR).Namespace(c.Namespace).Delete(context.TODO(), chaosId.Name, metav1.DeleteOptions{})
	case resource.ExperimentStressChaosKind:
		return c.DynamicClient.Resource(stressChaosGVR).Namespace(c.Namespace).Delete(context.TODO(), chaosId.Name, metav1.DeleteOptions{})
	}
	return fmt.Errorf("unknown chaos identifier kind: %s", chaosId.Spec.Kind())
}

func (c *ChaosClient) createPodFailureObject(opts resource.InjectChaosOptions, spec *resource.FailureExperimentSpec) (*ChaosIdentifier, error) {
	if len(opts.Selector) == 0 {
		return nil, fmt.Errorf("expect a non-empty pod selector")
	}
	if opts.Target != nil {
		return nil, fmt.Errorf("the opts.Target should be nil when create a failure injection")
	}

	failureObj := new(unstructured.Unstructured)
	if err := mergo.Merge(&failureObj.Object, runtime.DeepCopyJSON(defaultPodFailureSpec)); err != nil {
		return nil, err
	}
	failureObj.SetName(resource.GenResourceName("", "pod-failure-"))

	podSelector := c.buildPodSelector(opts.Selector)
	_ = unstructured.SetNestedField(failureObj.Object, podSelector.Pods, "spec", "selector", "pods")
	if spec.Duration != "" {
		_ = unstructured.SetNestedField(failureObj.Object, spec.Duration, "spec", "duration")
	}
	if spec.Scheduler != nil {
		_ = unstructured.SetNestedField(failureObj.Object, spec.Scheduler.Cron, "spec", "scheduler", "cron")
	}
	obj, err := c.DynamicClient.Resource(podChaosGVR).Namespace(c.Namespace).Create(context.TODO(), failureObj, metav1.CreateOptions{})
	if err != nil {
		return nil, err
	}
	return &ChaosIdentifier{Namespace: c.Namespace, Name: obj.GetName(), Spec: spec}, nil
}

func (c *ChaosClient) createNetworkPartitionObject(opts resource.InjectChaosOptions, spec *resource.NetworkPartitionSpec) (resource.ChaosIdentifier, error) {
	if len(opts.Selector) == 0 {
		return nil, fmt.Errorf("expect a non-empty pod selector")
	}
	if opts.Target == nil || len(*opts.Target) == 0 {
		return nil, fmt.Errorf("expect a non-empty target pod selector")
	}

	partitionObj := new(unstructured.Unstructured)
	if err := mergo.Merge(&partitionObj.Object, runtime.DeepCopyJSON(defaultNetworkPartitionSpec)); err != nil {
		return nil, err
	}
	partitionObj.SetName(resource.GenResourceName("", "network-partition-"))

	podSelector := c.buildPodSelector(opts.Selector)
	targetPodSelector := c.buildPodSelector(*opts.Target)
	_ = unstructured.SetNestedField(partitionObj.Object, podSelector.Pods, "spec", "selector", "pods")
	_ = unstructured.SetNestedField(partitionObj.Object, targetPodSelector.Pods, "spec", "target", "selector", "pods")
	if spec.Duration != "" {
		_ = unstructured.SetNestedField(partitionObj.Object, spec.Duration, "spec", "duration")
	}
	if spec.Direction != "" {
		_ = unstructured.SetNestedField(partitionObj.Object, spec.Direction, "spec", "direction")
	}
	if spec.Scheduler != nil {
		_ = unstructured.SetNestedField(partitionObj.Object, spec.Scheduler.Cron, "spec", "scheduler", "cron")
	}

	obj, err := c.DynamicClient.Resource(networkChaosGVR).Namespace(c.Namespace).Create(context.TODO(), partitionObj, metav1.CreateOptions{})
	if err != nil {
		return nil, err
	}
	return &ChaosIdentifier{Namespace: c.Namespace, Name: obj.GetName(), Spec: spec}, nil
}

func (c *ChaosClient) createNetworkLossObject(opts resource.InjectChaosOptions, spec *resource.NetworkLossSpec) (resource.ChaosIdentifier, error) {
	if len(opts.Selector) == 0 {
		return nil, fmt.Errorf("expect a non-empty pod selector")
	}
	lossObj := new(unstructured.Unstructured)
	if err := mergo.Merge(&lossObj.Object, runtime.DeepCopyJSON(defaultNetworkLossSpec)); err != nil {
		return nil, err
	}
	lossObj.SetName(resource.GenResourceName("", "network-loss-"))

	podSelector := c.buildPodSelector(opts.Selector)
	_ = unstructured.SetNestedField(lossObj.Object, podSelector.Pods, "spec", "selector", "pods")
	if opts.Target != nil && len(*opts.Target) > 0 {
		targetPodSelector := c.buildPodSelector(*opts.Target)
		_ = unstructured.SetNestedField(lossObj.Object, "to", "spec", "direction")
		_ = unstructured.SetNestedField(lossObj.Object, targetPodSelector.Pods, "spec", "target", "selector", "pods")
		_ = unstructured.SetNestedField(lossObj.Object, "all", "spec", "target", "mode")
	}
	if spec.Duration != "" {
		_ = unstructured.SetNestedField(lossObj.Object, spec.Duration, "spec", "duration")
	}
	if spec.Scheduler != nil {
		_ = unstructured.SetNestedField(lossObj.Object, spec.Scheduler.Cron, "spec", "scheduler", "cron")
	}
	_ = unstructured.SetNestedField(lossObj.Object, spec.Loss, "spec", "loss", "loss")
	_ = unstructured.SetNestedField(lossObj.Object, spec.Correlation, "spec", "loss", "correlation")

	obj, err := c.DynamicClient.Resource(networkChaosGVR).Namespace(c.Namespace).Create(context.TODO(), lossObj, metav1.CreateOptions{})
	if err != nil {
		return nil, err
	}

	return &ChaosIdentifier{Namespace: c.Namespace, Name: obj.GetName(), Spec: spec}, nil
}

func (c *ChaosClient) createNetworkDelayObject(opts resource.InjectChaosOptions, spec *resource.NetworkLatencySpec) (resource.ChaosIdentifier, error) {
	if len(opts.Selector) == 0 {
		return nil, fmt.Errorf("expect a non-empty pod selector")
	}
	delayObj := new(unstructured.Unstructured)
	if err := mergo.Merge(&delayObj.Object, runtime.DeepCopyJSON(defaultNetworkDelaySpec)); err != nil {
		return nil, err
	}
	delayObj.SetName(resource.GenResourceName("", "network-delay-"))

	podSelector := c.buildPodSelector(opts.Selector)
	_ = unstructured.SetNestedField(delayObj.Object, podSelector.Pods, "spec", "selector", "pods")
	if opts.Target != nil && len(*opts.Target) > 0 {
		targetPodSelector := c.buildPodSelector(*opts.Target)
		_ = unstructured.SetNestedField(delayObj.Object, "to", "spec", "direction")
		_ = unstructured.SetNestedField(delayObj.Object, targetPodSelector.Pods, "spec", "target", "selector", "pods")
		_ = unstructured.SetNestedField(delayObj.Object, "all", "spec", "target", "mode")
	}
	if spec.Duration != "" {
		_ = unstructured.SetNestedField(delayObj.Object, spec.Duration, "spec", "duration")
	}
	if spec.Scheduler != nil {
		_ = unstructured.SetNestedField(delayObj.Object, spec.Scheduler.Cron, "spec", "scheduler", "cron")
	}
	_ = unstructured.SetNestedField(delayObj.Object, spec.Latency, "spec", "delay", "latency")
	_ = unstructured.SetNestedField(delayObj.Object, spec.Jitter, "spec", "delay", "jitter")
	_ = unstructured.SetNestedField(delayObj.Object, spec.Correlation, "spec", "delay", "correlation")

	obj, err := c.DynamicClient.Resource(networkChaosGVR).Namespace(c.Namespace).Create(context.TODO(), delayObj, metav1.CreateOptions{})
	if err != nil {
		return nil, err
	}
	return &ChaosIdentifier{Namespace: c.Namespace, Name: obj.GetName(), Spec: spec}, nil
}

func (c *ChaosClient) createIODelayObject(opts resource.InjectChaosOptions, spec *resource.IODelaySpec) (resource.ChaosIdentifier, error) {
	if len(opts.Selector) == 0 {
		return nil, fmt.Errorf("expect a non-empty pod selector")
	}
	if opts.Target != nil {
		return nil, fmt.Errorf("the opts.Target should be nil when create a io delay injection")
	}

	delayObj := new(unstructured.Unstructured)
	if err := mergo.Merge(&delayObj.Object, runtime.DeepCopyJSON(defaultIODelaySpec)); err != nil {
		return nil, err
	}
	delayObj.SetName(resource.GenResourceName("", "io-delay-"))

	podSelector := c.buildPodSelector(opts.Selector)
	if err := unstructured.SetNestedField(delayObj.Object, podSelector.Pods, "spec", "selector", "pods"); err != nil {
		return nil, err
	}
	if spec.Duration != "" {
		if err := unstructured.SetNestedField(delayObj.Object, spec.Duration, "spec", "duration"); err != nil {
			return nil, err
		}
	}
	if spec.Scheduler != nil {
		if err := unstructured.SetNestedField(delayObj.Object, spec.Scheduler.Cron, "spec", "scheduler", "cron"); err != nil {
			return nil, err
		}
	}
	if err := unstructured.SetNestedField(delayObj.Object, spec.Delay, "spec", "delay"); err != nil {
		return nil, err
	}
	if err := unstructured.SetNestedField(delayObj.Object, spec.Percent, "spec", "percent"); err != nil {
		return nil, err
	}
	if err := unstructured.SetNestedField(delayObj.Object, spec.VolumePath, "spec", "volumePath"); err != nil {
		return nil, err
	}
	if err := unstructured.SetNestedField(delayObj.Object, spec.Path, "spec", "path"); err != nil {
		return nil, err
	}

	obj, err := c.DynamicClient.Resource(ioChaosGVR).Namespace(c.Namespace).Create(context.TODO(), delayObj, metav1.CreateOptions{})
	if err != nil {
		return nil, err
	}
	return &ChaosIdentifier{Namespace: c.Namespace, Name: obj.GetName(), Spec: spec}, nil
}

func (c *ChaosClient) createTimeChaosObject(opts resource.InjectChaosOptions, spec *resource.TimeChaosSpec) (resource.ChaosIdentifier, error) {
	if len(opts.Selector) == 0 {
		return nil, fmt.Errorf("expect a non-empty pod selector")
	}
	if opts.Target != nil {
		return nil, fmt.Errorf("the opts.Target should be nil when create a time chaos injection")
	}

	delayObj := new(unstructured.Unstructured)
	if err := mergo.Merge(&delayObj.Object, runtime.DeepCopyJSON(defaultTimeChaosSpec)); err != nil {
		return nil, err
	}
	delayObj.SetName(resource.GenResourceName("", "time-chaos-"))

	podSelector := c.buildPodSelector(opts.Selector)
	if err := unstructured.SetNestedField(delayObj.Object, podSelector.Pods, "spec", "selector", "pods"); err != nil {
		return nil, err
	}
	if spec.Duration != "" {
		if err := unstructured.SetNestedField(delayObj.Object, spec.Duration, "spec", "duration"); err != nil {
			return nil, err
		}
	}
	if spec.Scheduler != nil {
		if err := unstructured.SetNestedField(delayObj.Object, spec.Scheduler.Cron, "spec", "scheduler", "cron"); err != nil {
			return nil, err
		}
	}
	if err := unstructured.SetNestedField(delayObj.Object, spec.TimeOffset, "spec", "timeOffset"); err != nil {
		return nil, err
	}
	obj, err := c.DynamicClient.Resource(timeChaosGVR).Namespace(c.Namespace).Create(context.TODO(), delayObj, metav1.CreateOptions{})
	if err != nil {
		return nil, err
	}
	return &ChaosIdentifier{Namespace: c.Namespace, Name: obj.GetName(), Spec: spec}, nil
}

func (c *ChaosClient) createStressChaosObject(opts resource.InjectChaosOptions, spec *resource.StressChaosSpec) (resource.ChaosIdentifier, error) {
	if len(opts.Selector) == 0 {
		return nil, fmt.Errorf("expect a non-empty pod selector")
	}
	if opts.Target != nil {
		return nil, fmt.Errorf("the opts.Target should be nil when create a stress chaos injection")
	}

	stressObj := new(unstructured.Unstructured)
	if err := mergo.Merge(&stressObj.Object, runtime.DeepCopyJSON(defaultStressChaosSpec)); err != nil {
		return nil, err
	}
	stressObj.SetName(resource.GenResourceName("", "stress-chaos-"))

	podSelector := c.buildPodSelector(opts.Selector)
	if err := unstructured.SetNestedField(stressObj.Object, podSelector.Pods, "spec", "selector", "pods"); err != nil {
		return nil, err
	}
	if spec.Duration != "" {
		if err := unstructured.SetNestedField(stressObj.Object, spec.Duration, "spec", "duration"); err != nil {
			return nil, err
		}
	}
	if spec.Scheduler != nil {
		if err := unstructured.SetNestedField(stressObj.Object, spec.Scheduler.Cron, "spec", "scheduler", "cron"); err != nil {
			return nil, err
		}
	}

	if spec.Stressors.MemoryStressor != nil {
		if err := unstructured.SetNestedField(stressObj.Object, spec.Stressors.MemoryStressor.Workers, "spec", "stressors", "memory", "workers"); err != nil {
			return nil, err
		}
		if err := unstructured.SetNestedField(stressObj.Object, spec.Stressors.MemoryStressor.Size, "spec", "stressors", "memory", "size"); err != nil {
			return nil, err
		}
	} else {
		if err := unstructured.SetNestedField(stressObj.Object, nil, "spec", "stressors", "memory"); err != nil {
			return nil, err
		}
	}

	if spec.Stressors.CPUStressor != nil {
		if err := unstructured.SetNestedField(stressObj.Object, spec.Stressors.CPUStressor.Workers, "spec", "stressors", "cpu", "workers"); err != nil {
			return nil, err
		}
		if err := unstructured.SetNestedField(stressObj.Object, spec.Stressors.CPUStressor.Load, "spec", "stressors", "cpu", "load"); err != nil {
			return nil, err
		}
	} else {
		if err := unstructured.SetNestedField(stressObj.Object, nil, "spec", "stressors", "cpu"); err != nil {
			return nil, err
		}
	}

	obj, err := c.DynamicClient.Resource(stressChaosGVR).Namespace(c.Namespace).Create(context.TODO(), stressObj, metav1.CreateOptions{})
	if err != nil {
		return nil, err
	}
	return &ChaosIdentifier{Namespace: c.Namespace, Name: obj.GetName(), Spec: spec}, nil
}

type podSelector struct {
	Pods map[string]interface{}
}

func (c *ChaosClient) buildPodSelector(selector []resource.ChaosSelector) podSelector {
	pods := map[string]interface{}{}
	for _, s := range selector {
		podSelector := s.(*PodSelector)
		list, e := pods[podSelector.Namespace].([]interface{})
		if !e {
			list = []interface{}{}
		}
		list = append(list, podSelector.Name)
		pods[podSelector.Namespace] = list
	}
	return podSelector{pods}
}