package dwworker

import (
	"context"
	"crypto/tls"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"time"

	"github.com/pingcap/errors"
	"k8s.io/client-go/kubernetes"
	"sigs.k8s.io/controller-runtime/pkg/log"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
)

const (
	MaxConnsPerHost     = 1024
	MaxIdleConnsPerHost = 128
	IdleConnTimeout     = 30 * time.Second
)

type RequestDecorator func(req *http.Request) error

type APIClient interface {
	GetServiceInfo() (*cloudv1.DWWorkerStatus, error)
}

type APIClientWithKubeProxy struct {
	ctx context.Context

	client      kubernetes.Interface
	namespace   string
	serviceName string
	port        string
}

func newAPIClientWithKubeProxy(
	ctx context.Context, client kubernetes.Interface, namespace string, serviceName string, port int,
) *APIClientWithKubeProxy {
	return &APIClientWithKubeProxy{
		ctx:         ctx,
		client:      client,
		namespace:   namespace,
		serviceName: serviceName,
		port:        fmt.Sprint(port),
	}
}

func (g *APIClientWithKubeProxy) GetServiceInfo() (*cloudv1.DWWorkerStatus, error) {
	r := g.client.CoreV1().Services(g.namespace).ProxyGet("http", g.serviceName, g.port, "/info", nil)
	body, err := r.DoRaw(g.ctx)
	if err != nil {
		return nil, err
	}

	logger := log.FromContext(g.ctx)
	logger.Info("[api_client_kube_proxy] GET /info", "response", string(body))

	resp := &cloudv1.DWWorkerStatus{}
	err = json.Unmarshal(body, resp)
	if err != nil {
		return nil, err
	}
	return resp, nil
}

type APIClientWithClusterProxy struct {
	ctx               context.Context
	cli               *http.Client
	authDeco          RequestDecorator
	dwWorkerStatusURL string
}

func newAPIClientWithClusterProxy(
	ctx context.Context, clusterProxyBaseURL string,
	clusterName string, dwWorkerName string, authDeco RequestDecorator,
) *APIClientWithClusterProxy {
	tr := &http.Transport{
		MaxConnsPerHost:     MaxConnsPerHost,
		MaxIdleConnsPerHost: MaxIdleConnsPerHost,
		IdleConnTimeout:     IdleConnTimeout,
		/* #nosec */
		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
	}
	cli := &http.Client{Transport: tr}
	statusURL := &url.URL{
		Scheme: "http",
		Host:   clusterProxyBaseURL,
		Path:   fmt.Sprintf("/proxy/cluster/%s/v1alpha1/dwworker/%s/info", clusterName, dwWorkerName),
	}

	return &APIClientWithClusterProxy{
		ctx:               ctx,
		cli:               cli,
		authDeco:          authDeco,
		dwWorkerStatusURL: statusURL.String(),
	}
}

func (g *APIClientWithClusterProxy) GetServiceInfo() (*cloudv1.DWWorkerStatus, error) {
	logger := log.FromContext(g.ctx)

	req, err := http.NewRequestWithContext(g.ctx, http.MethodGet, g.dwWorkerStatusURL, nil)
	if err != nil {
		return nil, errors.Annotate(err, "failed to create request")
	}
	// authDeco may be nil in dev mode
	if g.authDeco != nil {
		if err := g.authDeco(req); err != nil {
			return nil, errors.Annotate(err, "failed to decorate request")
		}
	}
	resp, err := g.cli.Do(req)
	if err != nil {
		return nil, errors.Annotate(err, "failed to send request")
	}
	defer resp.Body.Close()
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, errors.Annotate(err, "failed to read response body")
	}
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("get http response body error, status code: %d, body: %s", resp.StatusCode, string(body))
	}

	logger.Info("[api_client_cluster_proxy] GET /info", "response", string(body))

	status := &cloudv1.DWWorkerStatus{}
	err = json.Unmarshal(body, status)
	if err != nil {
		return nil, errors.Annotate(err, "failed to unmarshal response body")
	}
	return status, nil
}