package managerserver

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net/http"
	"time"

	"github.com/pingcap/tidb/pkg/util"
	clientv3 "go.etcd.io/etcd/client/v3"
	"google.golang.org/grpc"
	"google.golang.org/grpc/keepalive"
)

const (
	autoidKey            = "/tidb/autoid/leader"
	autoIDOwnerPrompt    = "autoid"
	serverlessPDEndpoint = "serverless-cluster-pd.tidb-serverless.svc:2379"
)

func (s *server) getTiDBOwner(w http.ResponseWriter, r *http.Request) {
	logger := log.WithName("tidb/owner")
	logger.Info("req param", "req", r.URL)

	keyspaceID := r.URL.Query().Get("keyspace_id")
	ownerType := r.URL.Query().Get("owner_type")

	if len(keyspaceID) == 0 {
		logger.Error(nil, "keyspaceID is required")
		http.Error(w, "keyspaceID is required", http.StatusBadRequest)
		return
	}
	// init pd etcd client
	pdEtcdClient, err := s.getPDETCDClient()
	if err != nil {
		logger.Error(err, "get pd etcd client error", "error", err)
		http.Error(w, "get pd etcd client error", http.StatusInternalServerError)
		_, _ = io.WriteString(w, err.Error())
		return
	}

	ownerKey := getOwnerPath(ownerType, keyspaceID)
	logger.Info("get ownerKey:" + ownerKey)
	owner, err := s.getOwner(ownerKey, pdEtcdClient)
	if err != nil {
		logger.Error(err, "get owner error")
		http.Error(w, err.Error(), http.StatusInternalServerError)
		_, _ = io.WriteString(w, err.Error())
		return
	}

	_, err = io.WriteString(w, owner)
	if err != nil {
		logger.Error(err, "output error")
		http.Error(w, "output error", http.StatusInternalServerError)
		return
	}
}

func getOwnerPath(ownerType string, keyspaceID string) string {
	switch ownerType {
	case autoIDOwnerPrompt:
		return fmt.Sprintf("/keyspaces/tidb/%s"+autoidKey, keyspaceID)
	default:
		return ""
	}
}

func (s *server) getPDETCDClient() (*clientv3.Client, error) {
	tls, err := util.ToTLSConfig(s.cfg.ClusterCAPath, s.cfg.ClusterCertPath, s.cfg.ClusterKeyPath)
	if err != nil {
		return nil, err
	}
	pdEtcdClient, err := clientv3.New(clientv3.Config{
		Endpoints:   []string{serverlessPDEndpoint},
		TLS:         tls,
		DialTimeout: 5 * time.Second,
		DialOptions: []grpc.DialOption{
			grpc.WithKeepaliveParams(keepalive.ClientParameters{
				Time:                10 * time.Second,
				Timeout:             2 * time.Second,
				PermitWithoutStream: true,
			}),
		},
	})
	if err != nil {
		return nil, err
	}
	return pdEtcdClient, nil

}

func (s *server) getOwner(ownerPath string, pdEtcdClient *clientv3.Client) (string, error) {
	resp, err := pdEtcdClient.Get(context.Background(), ownerPath, clientv3.WithFirstCreate()...)
	if err != nil {
		return "", err
	}
	if len(resp.Kvs) == 0 {
		return "", nil
	}
	ownerID, _ := splitOwnerValues(resp.Kvs[0].Value)
	return string(ownerID), nil
}

func splitOwnerValues(val []byte) ([]byte, byte) {
	vals := bytes.Split(val, []byte("_"))
	var op byte
	if len(vals) == 2 {
		op = vals[1][0]
	}
	return vals[0], op
}