package portforward

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

	"github.com/pingcap/errors"
	"google.golang.org/grpc/resolver"
	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"
)

// RegisterClusterWithSchema registers a kubernetes cluster with a schema.
func RegisterClusterWithSchema(clientSet *kubernetes.Clientset,
	kubeConfig *restclient.Config, schema string) {
	resolver.Register(NewBuilder(clientSet, kubeConfig, schema))
}

// 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
}

type targetInfo struct {
	serviceName       string
	serviceNamespace  string
	port              string
	resolveByPortName bool
	useFirstPort      bool
}

func parseResolverTarget(target resolver.Target) (targetInfo, error) {
	var service, port, namespace string
	if target.URL.Host == "" {
		// kubernetes:///service.namespace:port
		service, port, namespace = splitServicePortNamespace(target.Endpoint())
	} else if target.URL.Port() == "" && target.Endpoint() != "" {
		// kubernetes://namespace/service:port
		service, port, _ = splitServicePortNamespace(target.Endpoint())
		namespace = target.URL.Hostname()
	} else {
		// kubernetes://service.namespace:port
		service, port, namespace = splitServicePortNamespace(target.URL.Host)
	}

	if service == "" {
		return targetInfo{}, fmt.Errorf("target %s must specify a service", &target.URL)
	}

	resolveByPortName := false
	useFirstPort := false
	if port == "" {
		useFirstPort = true
	} else if _, err := strconv.Atoi(port); err != nil {
		resolveByPortName = true
	}

	return targetInfo{
		serviceName:       service,
		serviceNamespace:  namespace,
		port:              port,
		resolveByPortName: resolveByPortName,
		useFirstPort:      useFirstPort,
	}, nil
}

func (ti targetInfo) String() string {
	return fmt.Sprintf("kubernetes://%s/%s:%s", ti.serviceNamespace, ti.serviceName, ti.port)
}

// NewBuilder creates a kubeBuilder which is used by grpc resolver.
func NewBuilder(clientSet *kubernetes.Clientset,
	kubeConfig *restclient.Config, schema string) resolver.Builder {
	return &kubeBuilder{
		clientSet:  clientSet,
		kubeConfig: kubeConfig,
		schema:     schema,
	}
}

// Build creates a new resolver for the given target.
//
// gRPC dial calls Build synchronously, and fails if the returned error is
// not nil.
func (b *kubeBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
	ti, err := parseResolverTarget(target)
	if err != nil {
		return nil, err
	}
	if ti.serviceNamespace == "" {
		ti.serviceNamespace = "nightly-ms"
	}
	ctx := context.Background()
	service, err := b.clientSet.CoreV1().Services(ti.serviceNamespace).Get(ctx, ti.serviceName, metav1.GetOptions{})
	if err != nil {
		return nil, errors.Trace(err)
	}
	portNum, err := strconv.ParseUint(ti.port, 10, 32)
	if err != nil {
		return nil, errors.Trace(err)
	}
	pod, err := getFirstPod(ctx, b.clientSet, service)
	if err != nil {
		return nil, errors.Trace(err)
	}
	remotePort, err := lookupContainerPortNumberByServicePort(service, pod, int32(portNum))
	if err != nil {
		return nil, errors.Trace(err)
	}
	localPort := getRandomPort()
	readyChannel := make(chan struct{})
	stopChannel := make(chan struct{}, 1)
	go func() {
		if err = forwardPorts(b.clientSet, b.kubeConfig, ti.serviceNamespace, pod.Name, int32(localPort), remotePort, stopChannel, readyChannel); err != nil {
			panic(err)
		}
	}()
	<-readyChannel
	cc.NewAddress([]resolver.Address{{Addr: fmt.Sprintf("localhost:%d", localPort)}})
	r := &kResolver{
		target: ti,
		cc:     cc,
	}
	return r, nil
}

// Scheme returns the scheme supported by this resolver.
// Scheme is defined at https://github.com/grpc/grpc/blob/master/doc/naming.md.
func (b *kubeBuilder) Scheme() string {
	return b.schema
}

type kubeBuilder struct {
	clientSet  *kubernetes.Clientset
	kubeConfig *restclient.Config
	schema     string
}

type kResolver struct {
	target      targetInfo
	cc          resolver.ClientConn
	stopChannel chan struct{}
}

// ResolveNow will be called by gRPC to try to resolve the target name again.
// It's just a hint, resolver can ignore this if it's not necessary.
func (k *kResolver) ResolveNow(resolver.ResolveNowOptions) {
}

// Close closes the resolver.
func (k *kResolver) Close() {
	k.stopChannel <- struct{}{}
}