// package cluster provides services and proxies related to the Kubernetes cluster.
package cluster

import (
	"bytes"
	"context"
	"crypto/tls"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net/http"
	"net/url"
	"strings"
	"time"

	"github.com/patrickmn/go-cache"
	"github.com/pingcap/errors"
	"github.com/pingcap/tiflow/pkg/etcd"
	"github.com/pingcap/tiflow/pkg/txnutil/gc"
	"github.com/tidbcloud/clusterproxy/resolve"
	"github.com/tidbcloud/clusterproxy/route"
	"github.com/tidbcloud/clusterproxy/route/router"
	"github.com/tidbcloud/clusterproxy/util"
	"github.com/tidbcloud/clusterproxy/util/cluster"
	pd "github.com/tikv/pd/client"
	"go.etcd.io/etcd/client/v3"
	"google.golang.org/grpc"
	"google.golang.org/grpc/backoff"
)

const (
	pdServiceFormat = "%s-pd"
	pdPort          = "2379"
)

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

var pdRoute = &route.Route{
	Name:    "pd",
	Path:    "pd",
	Version: route.V1,
	Schema: &route.Schema{
		PathRegex:       `.*`,
		RequiredHeaders: []string{},
		JSONBodyModel:   nil,
	},
	GetHandler: pdGetHandler,
}

type AuthType int

const (
	AuthTypeSQLUser AuthType = iota
	AuthTypeSharingCode

	dashboardLoginPath    = "/dashboard/api/user/login"
	tlsClientSecretFormat = "%s-cluster-client-secret"
)

var tokenCache = cache.New(20*time.Hour, 10*time.Minute)

type dashboardAuthPayload struct {
	Type     AuthType `json:"type"`
	Username string   `json:"username"`
	Password string   `json:"password"`
}

type passwordModel struct {
	Password string `json:"password"`
}

func pdGetHandler(c *route.Context, cl resolve.Cluster) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		scheme := "https"
		namespace := cl.Namespace
		pdServiceName := fmt.Sprintf(pdServiceFormat, cl.Name)
		tlsClientSecretName := fmt.Sprintf(tlsClientSecretFormat, cl.Name)
		config, err := util.GetTLSConfig(c.Cli, namespace, cl.Name, tlsClientSecretName)
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		if config == nil {
			scheme = "http"
		}
		target := cluster.ServiceURL(scheme, namespace, pdServiceName, pdPort)
		proxy := route.KubernetesServiceProxy(scheme, namespace, pdServiceName, pdPort)
		if strings.HasPrefix(r.URL.Path, "dashboard") {
			passwordFromParams := r.URL.Query().Get("password")
			password, err := base64.URLEncoding.DecodeString(passwordFromParams)
			if err != nil {
				http.Error(w, "failed to decode the password", http.StatusBadRequest)
				return
			}
			token, err := getToken(namespace, cl.Name, target, config, string(password))
			if err != nil {
				http.Error(w, err.Error(), http.StatusInternalServerError)
				return
			}
			r.Header.Set("Authorization", "Bearer "+token)
		}
		if strings.Contains(r.URL.Path, "clearcdcinfo") { // workaround, clear cdc info store in pd
			pdSecurityOption, err := util.GetPDSecurityOption(r.Context(), c.Cli, namespace, cl.Name, tlsClientSecretName)
			if err != nil {
				http.Error(w, fmt.Sprintf("get pd security option failed, %s", err.Error()), http.StatusInternalServerError)
				return
			}
			if err = clearCDCInfo(r.Context(), target, config, pdSecurityOption); err != nil {
				http.Error(w, fmt.Sprintf("clear cdc info failed, %s", err.Error()), http.StatusInternalServerError)
				return
			}
			return
		}
		proxy.Transport = &http.Transport{
			TLSClientConfig: config,
		}
		proxy.ServeHTTP(w, r)
	})
}

func getToken(namespace, name string, target *url.URL, config *tls.Config, password string) (string, error) {
	var token string
	// Cache tokens per clusters as a single namespace can host multiple PD services.
	key := fmt.Sprintf("%s/%s", namespace, name)
	if cachedToken, found := tokenCache.Get(key); found {
		return cachedToken.(string), nil
	}
	loginURL := target.String() + dashboardLoginPath
	username := "cloud_admin"
	if password == "" {
		username = "root"
	}
	payload := dashboardAuthPayload{
		Type:     AuthTypeSQLUser,
		Username: username,
		Password: password,
	}
	res, err := fetchToken(loginURL, payload, config)
	if err != nil {
		return "", errors.Annotatef(err, "pdaAPI fetch token for %s", username)
	}
	token = res.Token
	tokenCache.Set(key, token, cache.DefaultExpiration)
	return token, nil
}

type tokenResponse struct {
	Token  string    `json:"token"`
	Expire time.Time `json:"expire"`
}

func fetchToken(loginURL string, payload dashboardAuthPayload, tlsConfig *tls.Config) (*tokenResponse, error) {
	const defaultTimeout = 5 * time.Second
	client := &http.Client{
		Timeout:   defaultTimeout,
		Transport: &http.Transport{TLSClientConfig: tlsConfig},
	}

	data, err := json.Marshal(payload)
	if err != nil {
		return nil, err
	}

	res, err := client.Post(loginURL, "application/json", bytes.NewBuffer(data)) // nolint:bodyclose
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()
	body, err := ioutil.ReadAll(res.Body)
	if err != nil {
		return nil, err
	}
	if res.StatusCode == http.StatusOK {
		tokenResponse := &tokenResponse{}
		err = json.Unmarshal(body, tokenResponse)
		if err != nil {
			return nil, err
		}
		return tokenResponse, nil
	}
	return nil, fmt.Errorf("error response %d URL %s, response body: %s", res.StatusCode, loginURL, string(body[:]))
}

// workaround
func clearCDCInfo(ctx context.Context, pdUrl *url.URL, tlsConfig *tls.Config, pdSecurityOption *pd.SecurityOption) error {
	etcdClient, err := newEtcdClient(ctx, pdUrl, tlsConfig)
	if err != nil {
		return err
	}
	pdClient, err := newPdClient(ctx, pdUrl, tlsConfig, pdSecurityOption)
	if err != nil {
		return err
	}

	cdcEtcdClient := etcd.NewCDCEtcdClient(ctx, etcdClient)
	leases, err := cdcEtcdClient.GetCaptureLeases(ctx)
	if err != nil {
		return err
	}

	err = cdcEtcdClient.ClearAllCDCInfo(ctx)
	if err != nil {
		return err
	}

	err = cdcEtcdClient.RevokeAllLeases(ctx, leases)
	if err != nil {
		return err
	}

	err = gc.RemoveServiceGCSafepoint(ctx, pdClient, gc.CDCServiceSafePointID)
	if err != nil {
		return err
	}

	return nil
}

func newPdClient(ctx context.Context, pdUrl *url.URL, tlsConfig *tls.Config, pdSecurityOption *pd.SecurityOption) (pd.Client, error) {
	grpcTLSOption := util.GetGRPCConfig(tlsConfig)
	pdEndpoints := []string{pdUrl.String()}

	pdClient, err := pd.NewClientWithContext(
		ctx, pdEndpoints, *pdSecurityOption,
		// TODO(hi-rustin): add gRPC metrics to Options.
		// See also: https://github.com/pingcap/tiflow/pull/2341#discussion_r673032407.
		pd.WithGRPCDialOptions(
			grpcTLSOption,
			grpc.WithBlock(),
			grpc.WithConnectParams(grpc.ConnectParams{
				Backoff: backoff.Config{
					BaseDelay:  time.Second,
					Multiplier: 1.1,
					Jitter:     0.1,
					MaxDelay:   3 * time.Second,
				},
				MinConnectTimeout: 3 * time.Second,
			}),
		))
	if err != nil {
		return nil, errors.Annotatef(err,
			"fail to open PD client, please check pd address \"%s\"", pdUrl)
	}

	return pdClient, nil
}

func newEtcdClient(ctx context.Context, pdUrl *url.URL, tlsConfig *tls.Config) (*clientv3.Client, error) {
	grpcTLSOption := util.GetGRPCConfig(tlsConfig)

	return clientv3.New(clientv3.Config{
		Context:   ctx,
		Endpoints: []string{pdUrl.String()},
		TLS:       tlsConfig,
		// LogConfig:   &logConfig,
		DialTimeout: 30 * time.Second,
		// TODO(hi-rustin): add gRPC metrics to Options.
		// See also: https://github.com/pingcap/tiflow/pull/2341#discussion_r673018537.
		DialOptions: []grpc.DialOption{
			grpcTLSOption,
			grpc.WithBlock(),
			grpc.WithConnectParams(grpc.ConnectParams{
				Backoff: backoff.Config{
					BaseDelay:  time.Second,
					Multiplier: 1.1,
					Jitter:     0.1,
					MaxDelay:   3 * time.Second,
				},
				MinConnectTimeout: 3 * time.Second,
			}),
		},
	})
}