package clusteragent

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

	"github.com/go-resty/resty/v2"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
	"sigs.k8s.io/controller-runtime/pkg/log"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-agent/v1"
)

type client struct {
	client  *resty.Client
	svcAddr string
}

type Client interface {
	ExecCommand(ctx context.Context, req *apiv1.ExecuteCommandReq) (*apiv1.ExecuteCommandResp, error)
	PauseTC(ctx context.Context, req *apiv1.PauseTCReq) (*apiv1.PauseTCResp, error)
	ResumeTC(ctx context.Context, req *apiv1.ResumeTCReq) (*apiv1.ResumeTCResp, error)
	DeleteTCMember(ctx context.Context, clusterId, memberName string) (*apiv1.DeleteTCMemberResp, error)
	DeleteASTSAndResumeTC(ctx context.Context, req *apiv1.DeleteASTSAndResumeTCReq) (*apiv1.DeleteASTSAndResumeTCResp, error)
	Restart(ctx context.Context, req *apiv1.RestartReq) (*apiv1.RestartResp, error)
	ListNodes(ctx context.Context, clusterId string) (*apiv1.ListNodesResp, error)
	GetNode(ctx context.Context, clusterId, nodeName string) (*apiv1.Node, error)
	ListTCs(ctx context.Context, clusterId string) ([]*apiv1.TC, error)
	IsNodeReadyForScaleUpDown(ctx context.Context, req *apiv1.IsNodeReadyForScaleUpDownReq) (*apiv1.IsNodeReadyForScaleUpDownResp, error)
	ListPods(ctx context.Context, clusterId string) (*apiv1.ListPodsResp, error)
	GetPod(ctx context.Context, clusterId, podName string) (*apiv1.Pod, error)
	GetClusterTopology(ctx context.Context, clusterId string) (*apiv1.GetClusterTopologyResp, error)
	DeleteGKEHeterogeneousComputeClass(ctx context.Context, clusterId, name string) error
}

var (
	_ Client = &client{}
)

func NewClient(cluster *cloudv1.Cluster) (Client, error) {
	httpCli := resty.New()
	httpCli.SetTransport(otelhttp.NewTransport(httpCli.GetClient().Transport)).SetTimeout(time.Second * 5)
	clusterGatewayAddr := ""
	if cluster.Status.Gateway != nil {
		clusterGatewayAddr = cluster.Status.Gateway.Endpoint
	}
	if clusterGatewayAddr == "" {
		return nil, fmt.Errorf("clusterGateway endpoint is empty")
	}

	return &client{
		client:  httpCli,
		svcAddr: clusterGatewayAddr,
	}, nil
}

func (c *client) ExecCommand(ctx context.Context, req *apiv1.ExecuteCommandReq) (*apiv1.ExecuteCommandResp, error) {
	path := "/api/v1/clusters/{cluster_id}:executeCommand"
	var resp apiv1.ExecuteCommandResp
	_, err := c.call(ctx, resty.MethodPost, path, map[string]string{"cluster_id": req.ClusterId}, nil, req, &resp)
	return &resp, err
}

func (c *client) PauseTC(ctx context.Context, req *apiv1.PauseTCReq) (*apiv1.PauseTCResp, error) {
	path := "/api/v1/clusters/{cluster_id}/tcs:batch_pause"
	var resp apiv1.PauseTCResp
	_, err := c.call(ctx, resty.MethodPost, path, map[string]string{"cluster_id": req.ClusterId}, nil, req, &resp)
	return &resp, err
}

func (c *client) ResumeTC(ctx context.Context, req *apiv1.ResumeTCReq) (*apiv1.ResumeTCResp, error) {
	path := "/api/v1/clusters/{cluster_id}/tcs:batch_resume"
	var resp apiv1.ResumeTCResp
	_, err := c.call(ctx, resty.MethodPost, path, map[string]string{"cluster_id": req.ClusterId}, nil, req, &resp)
	return &resp, err
}

func (c *client) DeleteTCMember(ctx context.Context, clusterId, memberName string) (*apiv1.DeleteTCMemberResp, error) {
	path := "/api/v1/clusters/{cluster_id}:delete_tc_member"
	req := &apiv1.DeleteTCMemberReq{
		ClusterId:  clusterId,
		MemberName: memberName,
	}
	var resp apiv1.DeleteTCMemberResp
	_, err := c.call(ctx, resty.MethodPost, path, map[string]string{"cluster_id": clusterId}, nil, req, &resp)
	return &resp, err
}

func (c *client) ListTCs(ctx context.Context, clusterId string) ([]*apiv1.TC, error) {
	path := "/api/v1/clusters/{cluster_id}/tcs"
	var resp apiv1.ListTCsResp
	_, err := c.call(ctx, resty.MethodGet, path, map[string]string{"cluster_id": clusterId}, nil, nil, &resp)
	if err != nil {
		return nil, err
	}
	return resp.Tcs, nil
}

func (c *client) DeleteASTSAndResumeTC(ctx context.Context, req *apiv1.DeleteASTSAndResumeTCReq) (*apiv1.DeleteASTSAndResumeTCResp, error) {
	path := "/api/v1/clusters/{cluster_id}:delete_asts_and_resume_tc"
	var resp apiv1.DeleteASTSAndResumeTCResp
	_, err := c.call(ctx, resty.MethodPost, path, map[string]string{"cluster_id": req.ClusterId}, nil, req, &resp)
	return &resp, err
}

func (c *client) ListNodes(ctx context.Context, clusterId string) (*apiv1.ListNodesResp, error) {
	path := "/api/v1/clusters/{cluster_id}/nodes"
	var resp apiv1.ListNodesResp
	_, err := c.call(ctx, resty.MethodGet, path, map[string]string{"cluster_id": clusterId}, nil, nil, &resp)
	if err != nil {
		return nil, err
	}
	return &resp, nil
}

func (c *client) GetNode(ctx context.Context, clusterId, nodeName string) (*apiv1.Node, error) {
	path := "/api/v1/clusters/{cluster_id}/nodes/{node_name}"
	var resp apiv1.Node
	_, err := c.call(ctx, resty.MethodGet, path, map[string]string{"cluster_id": clusterId, "node_name": nodeName}, nil, nil, &resp)
	if err != nil {
		return nil, err
	}
	return &resp, nil
}

func (c *client) IsNodeReadyForScaleUpDown(ctx context.Context, req *apiv1.IsNodeReadyForScaleUpDownReq) (*apiv1.IsNodeReadyForScaleUpDownResp, error) {
	path := "/api/v1/clusters/{cluster_id}:is_node_ready_for_scale_up_down"
	var resp apiv1.IsNodeReadyForScaleUpDownResp
	queryParams := make([]string, 0)
	for _, ct := range req.ComponentTypes {
		queryParams = append(queryParams, ct.String())
	}
	_, err := c.call(ctx, resty.MethodGet, path, map[string]string{"cluster_id": req.ClusterId}, map[string][]string{"componentTypes": queryParams}, req, &resp)
	if err != nil {
		return nil, err
	}
	return &resp, nil
}

func (c *client) Restart(ctx context.Context, req *apiv1.RestartReq) (*apiv1.RestartResp, error) {
	path := "/api/v1/clusters/{cluster_id}:restart"
	var resp apiv1.RestartResp
	_, err := c.call(ctx, resty.MethodPost, path, map[string]string{"cluster_id": req.ClusterId}, nil, req, &resp)
	return &resp, err
}

func (c *client) ListPods(ctx context.Context, clusterId string) (*apiv1.ListPodsResp, error) {
	path := "/api/v1/clusters/{cluster_id}/pods"
	var resp apiv1.ListPodsResp
	_, err := c.call(ctx, resty.MethodGet, path, map[string]string{"cluster_id": clusterId}, nil, nil, &resp)
	if err != nil {
		return nil, err
	}
	return &resp, nil
}

func (c *client) GetPod(ctx context.Context, clusterId, podName string) (*apiv1.Pod, error) {
	path := "/api/v1/clusters/{cluster_id}/pods/{pod_name}"
	var resp apiv1.Pod
	_, err := c.call(ctx, resty.MethodGet, path, map[string]string{"cluster_id": clusterId, "pod_name": podName}, nil, nil, &resp)
	if err != nil {
		return nil, err
	}
	return &resp, nil
}

func (c *client) GetClusterTopology(ctx context.Context, clusterId string) (*apiv1.GetClusterTopologyResp, error) {
	path := "/api/v1/clusters/{cluster_id}/topology"
	var resp apiv1.GetClusterTopologyResp
	_, err := c.call(ctx, resty.MethodGet, path, map[string]string{"cluster_id": clusterId}, nil, nil, &resp)
	if err != nil {
		return nil, err
	}
	return &resp, nil
}

func (c *client) DeleteGKEHeterogeneousComputeClass(ctx context.Context, clusterId, name string) error {
	path := "/api/v1/clusters/{cluster_id}/gkeHeterogeneous/computeClasses/{name}"
	_, err := c.call(ctx, resty.MethodDelete, path, map[string]string{"cluster_id": clusterId, "name": name}, nil, nil, nil)
	return err
}

var (
	defaultProtojsonUnmarshalOptions = protojson.UnmarshalOptions{
		// enable this filed to enhance compatibility in where the upstream adds some new fields.
		DiscardUnknown: true,
	}
)

func (c *client) call(ctx context.Context, method, apiPath string, pathParams map[string]string, queryParams map[string][]string, body interface{}, output interface{}) (*resty.Response, error) {
	logger := log.FromContext(ctx, "apiPath", apiPath, "method", method)

	clusterId, ok := pathParams["cluster_id"]
	if !ok {
		return nil, fmt.Errorf("cluster_id not found in pathParams")
	}

	// Because cluster-agent is deployed behind of Cluster Gateway(served by kong), a prefix "/cluster-agent/{tidbNamespace}" is added to the path.
	apiPathPrefix := fmt.Sprintf("/cluster-agent/tidb%s", clusterId)
	url := fmt.Sprintf("%s%s%s", c.svcAddr, apiPathPrefix, apiPath)
	// url := fmt.Sprintf("http://localhost:9080%s", apiPath)

	request := c.client.R().EnableTrace().SetPathParams(pathParams).SetContext(ctx)
	if body != nil {
		request = request.SetBody(body)
	}

	for key, values := range queryParams {
		for _, v := range values {
			request = request.SetQueryParam(key, v)
		}
	}

	resp, err := request.Execute(method, url)
	if err != nil {
		return nil, fmt.Errorf("call clusterAgent failed: %w", err)
	}
	logger.Info("call clusterAgent", "url", url, "status", resp.StatusCode())
	if resp.StatusCode() != http.StatusOK {
		return resp, fmt.Errorf("call clusterAgent, status(%d)!=200,body=%s", resp.StatusCode(), string(resp.Body()))
	}
	if output != nil {
		err = defaultProtojsonUnmarshalOptions.Unmarshal(resp.Body(), output.(proto.Message))
		if err != nil {
			return resp, fmt.Errorf("unmarshal response failed: %s", err)
		}
	}
	return resp, nil
}