package k8s

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/url"
	"path"
	"strings"

	"github.com/pingcap/errors"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/tools/remotecommand"
	"k8s.io/client-go/util/exec"

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

// KubeClientSet supports common operations used by sdk; it keeps some clients and a namespace.
type KubeClientSet struct {
	Namespace     string
	kubeConfig    string
	clientConfig  *rest.Config
	ClientSet     kubernetes.Interface
	DynamicClient dynamic.Interface
}

// NewKubeClientFromKubeConfig creates a KubeClientSet instance from the namespace and kubeConfig
func NewKubeClientFromKubeConfig(namespace string, kubeConfig string) (*KubeClientSet, error) {
	var (
		err error
		k   = &KubeClientSet{}
	)
	if len(namespace) == 0 {
		return nil, errors.New("empty namespace")
	}
	k.Namespace = namespace
	k.kubeConfig = kubeConfig
	k.clientConfig, err = LoadConfig(kubeConfig)
	if err != nil {
		return nil, err
	}
	if k.ClientSet, err = kubernetes.NewForConfig(k.clientConfig); err != nil {
		return nil, err
	}
	if k.DynamicClient, err = dynamic.NewForConfig(k.clientConfig); err != nil {
		return nil, err
	}
	return k, nil
}

// NewKubeClientFromKubeConfigData creates a KubeClientSet instance from the namespace and kubeConfigData
func NewKubeClientFromKubeConfigData(namespace string, kubeConfigData []byte) (*KubeClientSet, error) {
	var (
		err error
		k   = &KubeClientSet{}
	)
	if len(namespace) == 0 {
		return nil, errors.New("empty namespace")
	}
	k.Namespace = namespace
	k.clientConfig, err = clientcmd.RESTConfigFromKubeConfig(kubeConfigData)
	if err != nil {
		return nil, err
	}
	if k.ClientSet, err = kubernetes.NewForConfig(k.clientConfig); err != nil {
		return nil, err
	}
	if k.DynamicClient, err = dynamic.NewForConfig(k.clientConfig); err != nil {
		return nil, err
	}
	return k, nil
}

// NewKubeClient creates a KubeClientSet instance
func NewKubeClient(namespace string, clientConfig *rest.Config, kubeConfig string, clientSet kubernetes.Interface, dynamicClient dynamic.Interface) *KubeClientSet {
	return &KubeClientSet{
		Namespace:     namespace,
		clientConfig:  clientConfig,
		kubeConfig:    kubeConfig,
		ClientSet:     clientSet,
		DynamicClient: dynamicClient,
	}
}

// ExecOptions passed to ExecWithOptions
type ExecOptions struct {
	Command       []string
	Namespace     string
	PodName       string
	ContainerName string
	Stdin         io.Reader
	CaptureStdout bool
	CaptureStderr bool
	// If false, whitespace in std{err,out} will be removed.
	PreserveWhitespace bool
}

// ExecWithOptions executes a command in the specified container,
// returning stdout, stderr, exit code(optional) and error. `options` allowed for
// additional parameters to be passed.
func (ku *KubeClientSet) ExecWithOptions(options ExecOptions) (string, string, *int, error) {
	const tty = false
	var exitCode *int

	req := ku.ClientSet.CoreV1().RESTClient().Post().
		Resource("pods").
		Name(options.PodName).
		Namespace(options.Namespace).
		SubResource("exec").
		Param("container", options.ContainerName)
	req.VersionedParams(&corev1.PodExecOptions{
		Container: options.ContainerName,
		Command:   options.Command,
		Stdin:     options.Stdin != nil,
		Stdout:    options.CaptureStdout,
		Stderr:    options.CaptureStderr,
		TTY:       tty,
	}, scheme.ParameterCodec)
	sugar.Infof("it should be noted that a long-running command will not be interrupted even the use case has ended. For more information, please refer to https://github.com/pingcap/test-infra/discussions/129")
	defer sugar.Sync() //nolint:errcheck

	var stdout, stderr bytes.Buffer
	err := execute("POST", req.URL(), ku.clientConfig, options.Stdin, &stdout, &stderr, tty)

	if err == nil {
		exitCode = &[]int{0}[0]
	} else if exitErr, ok := err.(exec.CodeExitError); ok {
		exitCode = &exitErr.Code
	}
	if options.PreserveWhitespace {
		return stdout.String(), stderr.String(), exitCode, err
	}
	return strings.TrimSpace(stdout.String()), strings.TrimSpace(stderr.String()), exitCode, err
}

// ExecCommandInContainerWithFullOutput executes a command in the
// specified container and return stdout, stderr, exit code(optional) and error
func (ku *KubeClientSet) ExecCommandInContainerWithFullOutput(podName, containerName string, cmd ...string) (string, string, *int, error) {
	return ku.ExecWithOptions(ExecOptions{
		Command:            cmd,
		Namespace:          ku.Namespace,
		PodName:            podName,
		ContainerName:      containerName,
		Stdin:              nil,
		CaptureStdout:      true,
		CaptureStderr:      true,
		PreserveWhitespace: false,
	})
}

var _ resource.RemoteCommanderInterruptible = &RemoteCommander{}

type RemoteCommander struct {
	stdinReader   *io.PipeReader
	stdinWriter   *io.PipeWriter
	errorChan     chan resource.RemoteCommandErr
	interruptible bool
}

func (j *RemoteCommander) Interrupt() error {
	if j.interruptible {
		if _, err := j.stdinWriter.Write([]byte{3, 4}); err != nil {
			return err
		}
		return j.stdinWriter.Close()
	}
	return fmt.Errorf("uninterruptible")
}

func (j *RemoteCommander) Wait() resource.RemoteCommandErr {
	err := <-j.errorChan
	return err
}

func newRemoteCommander(errChan chan resource.RemoteCommandErr, interruptible bool) *RemoteCommander {
	var r *io.PipeReader
	var w *io.PipeWriter
	// if interruptible, we need sends ctrl+c into tty, so a pipe is created.
	if interruptible {
		r, w = io.Pipe()
	}
	return &RemoteCommander{
		stdinReader:   r,
		stdinWriter:   w,
		interruptible: interruptible,
		errorChan:     errChan,
	}
}

func (ku *KubeClientSet) ExecCommandInContainerWithAsyncOutputInterruptible(stdout io.WriteCloser, podName, containerName string, cmd ...string) (resource.RemoteCommanderInterruptible, error) {
	var exitCode *int

	req := ku.ClientSet.CoreV1().RESTClient().Post().
		Resource("pods").
		Name(podName).
		Namespace(ku.Namespace).
		SubResource("exec").
		Param("container", containerName)
	req.VersionedParams(&corev1.PodExecOptions{
		Container: containerName,
		Command:   cmd,
		Stdin:     true,
		Stdout:    true,
		Stderr:    true,
		TTY:       true,
	}, scheme.ParameterCodec)
	sugar.Infof("it should be noted that a long-running command will not be interrupted even the use case has ended. For more information, please refer to https://github.com/pingcap/test-infra/discussions/129")
	defer sugar.Sync() //nolint:errcheck

	errChan := make(chan resource.RemoteCommandErr, 1)
	commander := newRemoteCommander(errChan, true)

	go func() {
		defer func() {
			err := stdout.Close()
			if err != nil {
				sugar.Error("close the stdout WriteCloser error", zap.Error(err))
			}
		}()
		err := execute("POST", req.URL(), ku.clientConfig, commander.stdinReader, stdout, stdout, true)
		if err == nil {
			exitCode = &[]int{0}[0]
		} else if exitErr, ok := err.(exec.CodeExitError); ok {
			exitCode = &exitErr.Code
		} else {
			sugar.Warnw("cannot get the exit code, return -1")
			exitCode = &[]int{-1}[0]
		}
		errChan <- resource.RemoteCommandErr{
			Err:      err,
			ExitCode: *exitCode,
		}
	}()

	return commander, nil
}

func (ku *KubeClientSet) ExecCommandInContainerWithAsyncOutput(stdout, stderr io.WriteCloser, podName, containerName string, cmd ...string) (resource.RemoteCommander, error) {
	var exitCode *int

	req := ku.ClientSet.CoreV1().RESTClient().Post().
		Resource("pods").
		Name(podName).
		Namespace(ku.Namespace).
		SubResource("exec").
		Param("container", containerName)
	req.VersionedParams(&corev1.PodExecOptions{
		Container: containerName,
		Command:   cmd,
		Stdin:     false,
		Stdout:    true,
		Stderr:    true,
		TTY:       false,
	}, scheme.ParameterCodec)
	sugar.Infof("it should be noted that a long-running command will not be interrupted even the use case has ended. For more information, please refer to https://github.com/pingcap/test-infra/discussions/129")
	defer sugar.Sync() //nolint:errcheck

	errChan := make(chan resource.RemoteCommandErr, 1)
	commander := newRemoteCommander(errChan, false)

	go func() {
		defer func() {
			err := stdout.Close()
			if err != nil {
				sugar.Error("close the stdout WriteCloser error", zap.Error(err))
			}
			err = stderr.Close()
			if err != nil {
				sugar.Error("close the stderr WriteCloser error", zap.Error(err))
			}
		}()
		err := execute("POST", req.URL(), ku.clientConfig, nil, stdout, stderr, false)
		if err == nil {
			exitCode = &[]int{0}[0]
		} else if exitErr, ok := err.(exec.CodeExitError); ok {
			exitCode = &exitErr.Code
		} else {
			sugar.Warnw("cannot get the exit code, return -1")
			exitCode = &[]int{-1}[0]
		}
		errChan <- resource.RemoteCommandErr{
			Err:      err,
			ExitCode: *exitCode,
		}
	}()

	return commander, nil
}

func (ku *KubeClientSet) WriteFileInContainer(podName string, containerName string, content io.Reader, filepath string) error {
	cmd := []string{"mkdir", "-p", path.Dir(filepath)}
	stdout, stderr, _, err := ku.ExecCommandInContainerWithFullOutput(podName, containerName, cmd...)
	if err != nil {
		return errors.Annotatef(err, "failed to create directory for file (stdout=%q,stderr=%q)", stdout, stderr)
	}
	stdout, stderr, _, err = ku.ExecWithOptions(ExecOptions{
		Command:            []string{"cp", "/dev/stdin", filepath},
		Namespace:          ku.Namespace,
		PodName:            podName,
		ContainerName:      containerName,
		Stdin:              content,
		CaptureStdout:      true,
		CaptureStderr:      true,
		PreserveWhitespace: false,
	})
	if err != nil {
		return errors.Annotatef(err, "failed to write file (stdout=%q,stderr=%q)", stdout, stderr)
	}
	return nil
}

// TarStreamFrom builds a tar streaming
// ref: https://github.com/kubernetes/kubernetes/blob/edb0a72cff0e43bab72a02cada8486d562ee1cd5/staging/src/k8s.io/kubectl/pkg/cmd/cp/cp.go#L347-L380
func (ku *KubeClientSet) TarStreamFrom(podName string, containerName string, path string) io.ReadCloser {
	r, w := io.Pipe()
	req := ku.ClientSet.CoreV1().RESTClient().Post().
		Resource("pods").
		Name(podName).
		Namespace(ku.Namespace).
		SubResource("exec").
		Param("container", containerName)
	req.VersionedParams(&corev1.PodExecOptions{
		Container: containerName,
		Command:   []string{"sh", "-c", "tar cf - " + path},
		Stdin:     false,
		Stdout:    true,
		Stderr:    false,
		TTY:       false,
	}, scheme.ParameterCodec)
	go func() {
		err := execute("POST", req.URL(), ku.clientConfig, nil, w, nil, false)
		if err != nil {
			_ = w.CloseWithError(err)
			return
		}
		w.Close()
	}()
	return r
}

func (ku *KubeClientSet) AddResourceSpecToConfigMap(resourceHandle apiresource.ResourceHandle, configMapName string, spec json.RawMessage) error {
	cm, err := ku.ClientSet.CoreV1().ConfigMaps(ku.Namespace).Get(context.TODO(), configMapName, metav1.GetOptions{})
	if err != nil {
		return err
	}

	cm.Data[resourceHandle.Name] = string(spec)
	_, err = ku.ClientSet.CoreV1().ConfigMaps(ku.Namespace).Update(context.TODO(), cm, metav1.UpdateOptions{})

	return err
}

func (ku *KubeClientSet) GetResourceSpecFromConfigMap(resourceHandle apiresource.ResourceHandle, configMapName string) (json.RawMessage, error) {
	cm, err := ku.ClientSet.CoreV1().ConfigMaps(ku.Namespace).Get(context.TODO(), configMapName, metav1.GetOptions{})
	if err != nil {
		return nil, err
	}

	spec := []byte(cm.Data[resourceHandle.Name])
	return spec, nil
}

func (ku *KubeClientSet) RemoveResourceSpecFromConfigMap(resourceHandle apiresource.ResourceHandle, configMapName string) error {
	cm, err := ku.ClientSet.CoreV1().ConfigMaps(ku.Namespace).Get(context.TODO(), configMapName, metav1.GetOptions{})
	if err != nil {
		return err
	}

	delete(cm.Data, resourceHandle.Name)
	_, err = ku.ClientSet.CoreV1().ConfigMaps(ku.Namespace).Update(context.TODO(), cm, metav1.UpdateOptions{})

	return err
}

func (ku *KubeClientSet) execCommandInPodWithFullOutput(podName string, cmd ...string) (string, string, *int, error) {
	pod, err := ku.ClientSet.CoreV1().Pods(ku.Namespace).Get(context.TODO(), podName, metav1.GetOptions{})
	if err != nil {
		return "", "", nil, err
	}
	return ku.ExecCommandInContainerWithFullOutput(podName, pod.Spec.Containers[0].Name, cmd...)
}

func (ku *KubeClientSet) execCommandInContainerWithAsyncOutputInterruptible(stdout io.WriteCloser, podName string, cmd ...string) (resource.RemoteCommanderInterruptible, error) {
	pod, err := ku.ClientSet.CoreV1().Pods(ku.Namespace).Get(context.TODO(), podName, metav1.GetOptions{})
	if err != nil {
		return nil, err
	}
	return ku.ExecCommandInContainerWithAsyncOutputInterruptible(stdout, podName, pod.Spec.Containers[0].Name, cmd...)
}

func (ku *KubeClientSet) execCommandInContainerWithAsyncOutput(stdout io.WriteCloser, stderr io.WriteCloser, podName string, cmd ...string) (resource.RemoteCommander, error) {
	pod, err := ku.ClientSet.CoreV1().Pods(ku.Namespace).Get(context.TODO(), podName, metav1.GetOptions{})
	if err != nil {
		return nil, err
	}
	return ku.ExecCommandInContainerWithAsyncOutput(stdout, stderr, podName, pod.Spec.Containers[0].Name, cmd...)
}

func (ku *KubeClientSet) ListPods(labelMap map[string]string) ([]string, error) {
	pods, err := ku.ClientSet.CoreV1().Pods(ku.Namespace).List(context.TODO(), metav1.ListOptions{LabelSelector: labels.SelectorFromSet(labelMap).String()})
	if err != nil {
		return nil, err
	}
	var podNames []string
	for _, pod := range pods.Items {
		podNames = append(podNames, pod.Name)
	}
	return podNames, nil
}

func (ku *KubeClientSet) DeletePod(name string) error {
	return errors.Trace(ku.ClientSet.CoreV1().Pods(ku.Namespace).Delete(context.TODO(), name, metav1.DeleteOptions{}))
}

func execute(method string, url *url.URL, config *rest.Config, stdin io.Reader, stdout, stderr io.Writer, tty bool) error {
	exec, err := remotecommand.NewSPDYExecutor(config, method, url)
	if err != nil {
		return err
	}
	return exec.Stream(remotecommand.StreamOptions{
		Stdin:  stdin,
		Stdout: stdout,
		Stderr: stderr,
		Tty:    tty,
	})
}