package portforward

import (
	"context"
	"fmt"
	"net"
	"net/http"
	"os"
	"strings"

	"github.com/pingcap/errors"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/client-go/kubernetes"
	restclient "k8s.io/client-go/rest"
	"k8s.io/client-go/tools/portforward"
	"k8s.io/client-go/transport/spdy"
)

// getFirstPod returns the first pod of a service.
func getFirstPod(ctx context.Context, clientSet *kubernetes.Clientset, service *corev1.Service) (*corev1.Pod, error) {
	// labels.Parser
	set := labels.Set(service.Spec.Selector)

	pods, err := clientSet.CoreV1().Pods(service.Namespace).List(ctx, metav1.ListOptions{
		LabelSelector: set.AsSelector().String()})
	if err != nil {
		return nil, errors.Trace(err)
	}
	if len(pods.Items) == 0 {
		return nil, errors.New("no pod found")
	}
	pod := pods.Items[0]
	return &pod, nil
}

// lookupContainerPortNumberByServicePort find containerPort number by service port number
func lookupContainerPortNumberByServicePort(svc *corev1.Service, pod *corev1.Pod, port int32) (int32, error) {
	for _, svcPort := range svc.Spec.Ports {
		if svcPort.Port != port {
			continue
		}
		if svc.Spec.ClusterIP == corev1.ClusterIPNone {
			return port, nil
		}
		if svcPort.TargetPort.Type == intstr.Int {
			if svcPort.TargetPort.IntValue() == 0 {
				// targetPort is omitted, and the IntValue() would be zero
				return svcPort.Port, nil
			}
			return int32(svcPort.TargetPort.IntValue()), nil
		}
		return lookupContainerPortNumberByName(pod, svcPort.TargetPort.String())
	}
	return port, errors.New(fmt.Sprintf("Service %s does not have a service port %d", svc.Name, port))
}

// lookupContainerPortNumberByName find containerPort number by its named port name
func lookupContainerPortNumberByName(pod *corev1.Pod, name string) (int32, error) {
	for _, ctr := range pod.Spec.Containers {
		for _, ctrportspec := range ctr.Ports {
			if ctrportspec.Name == name {
				return ctrportspec.ContainerPort, nil
			}
		}
	}
	for _, ctr := range pod.Spec.InitContainers {
		for _, ctrportspec := range ctr.Ports {
			if ctrportspec.Name == name {
				return ctrportspec.ContainerPort, nil
			}
		}
	}
	return int32(-1), errors.New(fmt.Sprintf("Pod '%s' does not have a named port '%s'", pod.Name, name))
}

func forwardPorts(clientSet *kubernetes.Clientset, kubeConfig *restclient.Config,
	namespace string,
	podName string, localPort, remotePort int32, stopChan <-chan struct{}, readyChan chan struct{}) error {
	req := clientSet.CoreV1().RESTClient().Post().
		Resource("pods").
		Namespace(namespace).
		Name(podName).
		SubResource("portforward")
	transport, upgrader, err := spdy.RoundTripperFor(kubeConfig)
	if err != nil {
		return errors.Trace(err)
	}
	dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, "POST", req.URL())
	fw, err := portforward.NewOnAddresses(dialer, []string{"0.0.0.0"},
		[]string{fmt.Sprintf("%d:%d", localPort, remotePort)}, stopChan, readyChan, os.Stdout, os.Stderr)
	if err != nil {
		return err
	}
	return fw.ForwardPorts()
}

func splitServicePortNamespace(hpn string) (service, port, namespace string) {
	service = hpn
	colon := strings.LastIndexByte(service, ':')
	if colon != -1 {
		service, port = service[:colon], service[colon+1:]
	}

	// we want to split into the service name, namespace, and whatever else is left
	// this will support fully qualified service names, e.g. {service-name}.<namespace>.svc.<cluster-domain-name>.
	// Note that since we lookup the endpoints by service name and namespace, we don't care about the
	// cluster-domain-name, only that we can parse out the service name and namespace properly.
	parts := strings.SplitN(service, ".", 3)
	if len(parts) >= 2 {
		service, namespace = parts[0], parts[1]
	}

	return
}

func getRandomPort() int {
	l, err := net.Listen("tcp", ":0")
	if err != nil {
		panic(err)
	}
	port := l.Addr().(*net.TCPAddr).Port
	defer func() {
		_ = l.Close()
	}()
	return port
}