package gotty

import (
	"context"
	"flag"
	"fmt"
	"net/http"
	"strings"
	"time"

	"github.com/pingcap/log"
	"github.com/tidbcloud/clusterproxy/resolve"
	"github.com/tidbcloud/clusterproxy/route"
	"github.com/tidbcloud/clusterproxy/route/router"
	"github.com/tidbcloud/clusterproxy/util/cluster"
	"go.uber.org/zap"
	"google.golang.org/protobuf/proto"
	batchv1 "k8s.io/api/batch/v1"
	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/client-go/kubernetes"
)

const (
	gottyPort = "8080"
)

var (
	gottyImage = flag.String("gotty-image", "gcr.io/pingcap-public/dbaas/gotty:20201216", "image of gotty")
	gottyCmd   = flag.String("gotty-cmd", "/gotty -w --max-connection 1 --timeout 60 --reconnect /client-loop", "start command of gotty")
)

var gottyWaitTimeout = time.Second * 30

func init() {
	router.RegisterRoute(gottyRoute)
}

var gottyRoute = &route.Route{
	Name:    "terminal",
	Path:    "terminal",
	Version: route.V1,
	Schema: &route.Schema{
		PathRegex:       `^.*$`,
		JSONBodyModel:   nil,
		RequiredHeaders: []string{"x-user-id"},
	},
	GetHandler: gottyGetHandler,
}

func gottyGetHandler(c *route.Context, cl resolve.Cluster) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		userId := r.Header.Get("x-user-id")
		if userId == "" {
			http.Error(w, "Required header x-user-id not found", http.StatusBadRequest)
			return
		}
		paths := strings.Split(r.URL.Path, "/")
		if len(paths) == 0 {
			http.Error(w, "Path does not contain enough information", http.StatusBadRequest)
			return
		}
		user := paths[0]
		namespace := cl.Namespace

		svc := fmt.Sprintf("shell-%s-%s-%s", cl.Name, userId, user)

		var ready bool
		timeout := time.After(gottyWaitTimeout)
		for !ready {
			ready = runGotty(c.Cli, namespace, cl.Name, userId, user)
			if !ready {
				/* serverless: retry inner gateway */
				ready = runGottyWithGW(c.Cli, namespace, cl.Name, userId, user)
				if !ready {
					time.Sleep(time.Second)
				} else {
					/* serverless: with shared namespace */
					namespace = "tidb-pool"
				}
			}
			select {
			case <-timeout:
				log.Error("run gotty timeout")
				w.WriteHeader(http.StatusInternalServerError)
				return
			default:
			}
		}

		r.URL.Path = strings.TrimPrefix(r.URL.Path, user)

		route.KubernetesServiceProxy("http", namespace, svc, gottyPort).ServeHTTP(w, r)
	})
}

func runGottyWithGW(cli kubernetes.Interface, namespace, name, userId, user string) bool {
	userName := strings.TrimPrefix(namespace, "free-tier-") + "." + strings.TrimPrefix(name, "db") + "." + user
	dsn := fmt.Sprintf("ti://%s@tidb-gateway-inner.tidb-admin.svc.cluster.local:4000/", userName)
	svc := fmt.Sprintf("shell-%s-%s-%s", name, userId, user)
	/* serverless: with shared namespace */
	namespace = "tidb-pool"
	if err := startGottyIfNotExist(cli, namespace, svc, dsn); err != nil {
		log.Error("failed start gotty", zap.Error(err))
		return false
	}
	_, port, err := getServiceAddress(cli, namespace, svc)
	if err != nil {
		log.Error("failed get service port", zap.Error(err))
		return false
	}
	resp, err := http.Get(fmt.Sprintf("%s://%s.%s:%d", "http", svc, namespace, port))
	if err != nil {
		log.Error("failed http get", zap.Error(err))
		return false
	}
	defer resp.Body.Close()
	return resp.StatusCode == http.StatusOK
}

func getServiceAddress(cli kubernetes.Interface, namespace, serviceName string) (ip string, port int32, err error) {
	ctx := context.TODO()

	var svc *v1.Service
	if svc, err = cli.CoreV1().Services(namespace).Get(ctx, serviceName, metav1.GetOptions{}); err != nil {
		return
	}
	for _, svcPort := range svc.Spec.Ports {
		if port == 0 || svcPort.Port < port {
			port = svcPort.Port
		}
	}
	if svc.Spec.ClusterIP == "" {
		err = fmt.Errorf("service %s does not have a cluster IP", serviceName)
		return
	}

	var ep *v1.Endpoints
	if ep, err = cli.CoreV1().Endpoints(namespace).Get(ctx, serviceName, metav1.GetOptions{}); err != nil {
		return
	}
	ready := false
	if len(ep.Subsets) > 0 {
		for _, subset := range ep.Subsets {
			if len(subset.Addresses) > 0 {
				ip = subset.Addresses[0].IP
				ready = true
				break
			}
		}
	}
	if !ready {
		err = fmt.Errorf("service %s backend is not ready", serviceName)
	}
	return
}

func runGotty(cli kubernetes.Interface, namespace, name, userId, user string) bool {
	serviceName := fmt.Sprintf(cluster.TiDBNameFormat, name)
	ip, _, err := getServiceAddress(cli, namespace, serviceName)
	if err != nil {
		log.Error("failed get service ip", zap.Error(err))
		return false
	}
	// Use another name rather than "gotty". See https://github.com/yudai/gotty/pull/230
	svc := fmt.Sprintf("shell-%s-%s-%s", name, userId, user)
	dsn := fmt.Sprintf("ti://%s@%s:4000/", user, ip)
	if err := startGottyIfNotExist(cli, namespace, svc, dsn); err != nil {
		log.Error("failed start gotty", zap.Error(err))
		return false
	}
	_, port, err := getServiceAddress(cli, namespace, svc)
	if err != nil {
		log.Error("failed get service port", zap.Error(err))
		return false
	}
	resp, err := http.Get(fmt.Sprintf("%s://%s.%s:%d", "http", svc, namespace, port))
	if err != nil {
		log.Error("failed http get", zap.Error(err))
		return false
	}
	defer resp.Body.Close()
	return resp.StatusCode == http.StatusOK
}

func startGottyIfNotExist(cli kubernetes.Interface, namespace, jobName, dsn string) error {
	ctx := context.TODO()

	cmds := strings.Split(*gottyCmd, " ")
	cmds = append(cmds, dsn)
	labels := map[string]string{
		"app":      "gotty",
		"instance": jobName,
	}

	job, err := cli.BatchV1().Jobs(namespace).Get(ctx, jobName, metav1.GetOptions{})
	if err != nil && !errors.IsNotFound(err) {
		return err
	}
	if err == nil && job.Status.CompletionTime != nil {
		policy := metav1.DeletePropagationBackground
		if err := cli.BatchV1().Jobs(namespace).Delete(ctx, job.Name, metav1.DeleteOptions{
			PropagationPolicy:  &policy,
			GracePeriodSeconds: proto.Int64(0)}); err != nil {
			return err
		}
	}

	if _, err := cli.BatchV1().Jobs(namespace).Create(ctx, &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{Name: jobName},
		Spec: batchv1.JobSpec{
			TTLSecondsAfterFinished: proto.Int32(120),
			Completions:             proto.Int32(1),
			BackoffLimit:            proto.Int32(1),
			Template: v1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: labels,
				},
				Spec: v1.PodSpec{
					RestartPolicy: v1.RestartPolicyNever,
					Containers: []v1.Container{{
						Name:  "gotty",
						Image: *gottyImage,
						Ports: []v1.ContainerPort{{
							ContainerPort: 8080,
							Name:          "websocket",
							Protocol:      v1.ProtocolTCP,
						}},
						Command: cmds,
						Resources: v1.ResourceRequirements{
							Requests: v1.ResourceList{
								v1.ResourceCPU:    resource.MustParse("0.1"),
								v1.ResourceMemory: resource.MustParse("300Mi"),
							},
						},
					}},
				},
			},
		},
	}, metav1.CreateOptions{}); err != nil && !errors.IsAlreadyExists(err) {
		return err
	}

	if _, err := cli.CoreV1().Services(namespace).Create(ctx, &v1.Service{
		ObjectMeta: metav1.ObjectMeta{Name: jobName},
		Spec: v1.ServiceSpec{
			Selector: labels,
			Ports: []v1.ServicePort{{
				Name:       "websocket",
				Port:       8080,
				TargetPort: intstr.FromInt(8080),
			}},
		},
	}, metav1.CreateOptions{}); err != nil && !errors.IsAlreadyExists(err) {
		return err
	}
	return nil
}