package k8s

import (
	"context"
	_ "embed"
	"encoding/json"
	"fmt"
	"net/url"
	"os"
	"time"

	"github.com/hashicorp/go-multierror"
	"github.com/imdario/mergo"
	"github.com/pingcap/errors"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/watch"

	apiresource "github.com/pingcap/test-infra/common/model/resource"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
)

var (
	_ resource.Client = DMClusterClient{}

	//go:embed assets/dm-cluster.yaml
	dMClusterSpec        []byte
	defaultDMClusterSpec map[string]interface{}
)

func init() {
	mustUnmarshalYaml(dMClusterSpec, &defaultDMClusterSpec)

	core.RegisterProvider(resource.FQDN(DMClusterClient{}),
		func(handle apiresource.ResourceHandle) core.ResourceGetter { return &dmGetter{handle: handle} })
}

type dmGetter struct{ handle apiresource.ResourceHandle }

func (d *dmGetter) Get() (core.Resource, error) {
	k, err := NewKubeClientFromKubeConfig(d.handle.Testbed, os.Getenv("KUBECONFIG"))
	if err != nil {
		return nil, err
	}
	return NewDMClusterClient(k).Get(d.handle)
}

type DMClusterClient struct {
	*KubeClientSet
}

func NewDMClusterClient(k *KubeClientSet) DMClusterClient {
	return DMClusterClient{k}
}

func (d DMClusterClient) Create(spec json.RawMessage) (core.Resource, error) {
	dcr, dmRes, err := d.buildDMClusterObject(spec)
	if err != nil {
		return nil, errors.Annotate(err, "build dm-cluster resource object")
	}

	_, err = d.DynamicClient.Resource(dmGVR).Namespace(d.Namespace).Create(context.TODO(), dcr, metav1.CreateOptions{})
	if err != nil {
		return nil, errors.Annotate(err, "create dm-cluster resource")
	}

	dmResource := &DMCluster{
		KubeClientSet: *d.KubeClientSet,
		DMCluster: apiresource.DMCluster{
			ResourceHandle: dmRes.ResourceHandle,
		},
	}
	dmResource.Provider = d.FQDN()
	if err := dmResource.Sync(); err != nil {
		return nil, err
	}
	return dmResource, nil
}

func (d DMClusterClient) Get(handle apiresource.ResourceHandle) (core.Resource, error) {
	dmResource := &DMCluster{
		KubeClientSet: *d.KubeClientSet,
		DMCluster: apiresource.DMCluster{
			ResourceHandle: handle,
		},
	}
	dmResource.Provider = d.FQDN()
	if err := dmResource.Sync(); err != nil {
		return nil, err
	}
	return dmResource, nil
}

func (d DMClusterClient) GetEndpoints(_ apiresource.ResourceHandle) (apiresource.ResourceEndpoints, error) {
	return apiresource.ResourceEndpoints{}, nil
}

func (d DMClusterClient) Delete(handle apiresource.ResourceHandle) error {
	watcher, err := d.DynamicClient.Resource(dmGVR).Namespace(d.Namespace).Watch(context.TODO(), metav1.ListOptions{
		LabelSelector: labels.Set{TestInfraName: handle.Name}.AsSelector().String(),
	})
	if err != nil {
		return err
	}
	ch := watcher.ResultChan()

	err = d.DynamicClient.Resource(dmGVR).Namespace(d.Namespace).Delete(context.TODO(), handle.Name, metav1.DeleteOptions{
		PropagationPolicy: &[]metav1.DeletionPropagation{metav1.DeletePropagationForeground}[0],
	})
	if err != nil {
		return err
	}
	// wait for dc deleted before delete pvcs
waitLoop:
	for {
		select {
		case event := <-ch:
			switch event.Type {
			case watch.Deleted:
				break waitLoop
			}
		case <-time.After(waitForDeletionTimeout):
			return fmt.Errorf("waiting for dmcluster %s/%s deleted timeout", d.Namespace, handle.Name)
		}
	}

	// also need to  delete dc's pvc
	err = d.ClientSet.CoreV1().PersistentVolumeClaims(d.Namespace).DeleteCollection(context.TODO(), metav1.DeleteOptions{}, metav1.ListOptions{
		LabelSelector: labels.Set{"app.kubernetes.io/instance": handle.Name}.AsSelector().String(),
	})

	if err != nil {
		return errors.Annotate(err, "delete persistent volume claims of "+handle.Name)
	}
	return nil
}

func (d DMClusterClient) ReserveEnv(handle apiresource.ResourceHandle, action apiresource.GcReserveEnvAction) error {
	return nil
}

func (d DMClusterClient) Artifact(handle apiresource.ResourceHandle, action apiresource.GcArtifactAction, archiver resource.ArchiverFunc) (map[string]*url.URL, error) {
	var errs error
	var urlMap = make(map[string]*url.URL)
	pods, err := d.ClientSet.CoreV1().Pods(d.Namespace).List(context.TODO(), metav1.ListOptions{
		LabelSelector: labels.Set{"app.kubernetes.io/instance": handle.Name}.AsSelector().String(),
	})
	if err != nil {
		return nil, err
	}
	for _, pod := range pods.Items {
		if pod.Status.Phase == corev1.PodRunning {
			component := pod.Labels["app.kubernetes.io/component"]
			switch component {
			case "dm-master", "dm-worker":
				path := map[string]string{
					"dm-master": "/var/lib/dm-master/log/",
					"dm-worker": "/var/lib/dm-worker/log/",
				}[component]
				tarFilesPrefix := path[1:]
				cUrls, err := untarAll(d.TarStreamFrom(pod.Name, component, path), archiver, pod.Name, tarFilesPrefix)
				if err != nil {
					errs = multierror.Append(errs, err)
					continue
				}
				for component, url := range cUrls {
					urlMap[component] = url
				}
			default:
				continue
			}
		}
	}
	return urlMap, errs
}

func (d DMClusterClient) FQDN() string {
	return resource.FQDN(d)
}

func newDefaultDMClusterSpec() apiresource.DMClusterSpec {
	return apiresource.DMClusterSpec{
		Master: apiresource.MasterSpec{
			Replicas:  1,
			BaseImage: "",
		},
		Worker: &apiresource.WorkerSpec{
			Replicas:  1,
			BaseImage: "",
		},
		Version: "",
		Labels: map[string]string{
			"": "",
		},
	}
}

// buildDMClusterObject
// params should be the apiresource.DMCluster with some extension fields
func (d DMClusterClient) buildDMClusterObject(params json.RawMessage) (*unstructured.Unstructured, *apiresource.DMCluster, error) {
	var (
		rawSpec map[string]interface{}
		dcr     = new(unstructured.Unstructured)                          // the dm-cluster CR object
		dmcRes  = &apiresource.DMCluster{Spec: newDefaultDMClusterSpec()} // the api resource handle object
	)
	// unmarshal rawSpec obj
	var rawObj map[string]interface{}
	if err := json.Unmarshal(params, &rawObj); err != nil {
		return nil, nil, errors.Trace(err)
	}

	if spec, ok := rawObj["spec"].(map[string]interface{}); ok {
		rawSpec = spec
	}
	// unmarshal resource handle
	if err := json.Unmarshal(params, dmcRes); err != nil {
		return nil, nil, errors.Trace(err)
	}

	if err := mergo.Merge(&dcr.Object, runtime.DeepCopyJSON(defaultDMClusterSpec)); err != nil {
		return nil, nil, errors.Trace(err)
	}
	// set the dm-cluster CR object
	dcr.SetName(dmcRes.Name)
	dcr.SetNamespace(d.Namespace)
	// add the testbed to dcr's ownerReference
	if owner, err := GetOwnerReferenceFromTestbed(d.KubeClientSet, dmcRes.Testbed); err != nil {
		return nil, nil, err
	} else {
		dcr.SetOwnerReferences([]metav1.OwnerReference{*owner})
	}
	spec := dcr.Object["spec"].(map[string]interface{})
	if err := mergo.Merge(&spec, rawSpec, mergo.WithOverride); err != nil {
		return nil, nil, errors.Trace(err)
	}
	// rawObj labels
	ls := dcr.GetLabels()
	if err := mergo.Merge(&ls, map[string]string{TestInfraName: dmcRes.Name}, mergo.WithOverride); err != nil {
		return nil, nil, errors.Trace(err)
	}
	dcr.SetLabels(ls)
	return dcr, dmcRes, nil
}