package clusteragentcli

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

	"github.com/go-resty/resty/v2"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
	"k8s.io/utils/pointer"

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

type Client interface {
	CreateCertificate(ctx context.Context, req *apiv1.CreateCertificateReq) (*apicommon.Certificate, error)
}

type TenantPlanInfo struct {
	TenantID      uint   `json:"tenant_id"`
	EffectivePlan string `json:"effective_plan"`
}

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

func NewClient(addr string, token string) Client {
	httpCli := resty.New()
	httpCli.SetTransport(otelhttp.NewTransport(httpCli.GetClient().Transport)).SetTimeout(time.Second * 600)

	return &client{client: httpCli, svcAddr: addr, token: token}
}

func (c *client) CreateCertificate(ctx context.Context, req *apiv1.CreateCertificateReq) (*apicommon.Certificate, error) {
	path := "/api/v1/clusters/{cluster_id}/certificates"
	var cert apicommon.Certificate
	_, err := c.call(ctx, resty.MethodPost, path, map[string]string{"cluster_id": req.ClusterId}, nil, req, &cert)
	return &cert, 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 := tilog.WithContextL(ctx).With(zap.String("apiPath", apiPath), zap.String("method", method))

	clusterId, ok := pathParams["cluster_id"]
	if !ok {
		return nil, fmt.Errorf("cluster_id not found in pathParams")
	}
	agentAddr, err := c.getClusterAgentAddr(ctx, clusterId)
	if err != nil {
		return nil, err
	}
	// 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", agentAddr, apiPathPrefix, 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)
		}
	}

	request = request.SetAuthToken(c.token)

	resp, err := request.Execute(method, url)
	if err != nil {
		return nil, fmt.Errorf("call clusterAgent failed: %w", err)
	}
	logger.Info("call clusterAgent", zap.Int("status", resp.StatusCode()))
	if resp.StatusCode() != http.StatusOK {
		return resp, fmt.Errorf("call clusterAgent, status(%d)!=200", resp.StatusCode())
	}
	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
}

func (c *client) getClusterAgentAddr(ctx context.Context, clusterId string) (string, error) {
	clusterIdInt, err := strconv.ParseUint(clusterId, 10, 64)
	if err != nil {
		return "", fmt.Errorf("parse clusterId failed: %w", err)
	}
	resp, err := grpcli.GetClusterByID(ctx, clusterIdInt)
	if err != nil {
		return "", fmt.Errorf("get cluster failed: %w", err)
	}

	if resp.Attributes != nil && resp.Attributes.UseNewDedicatedArch != nil {
		if *resp.Attributes.UseNewDedicatedArch {
			return fmt.Sprintf("%s/proxy/clusters/%s", c.svcAddr, clusterId), nil
		}
	}

	addr := pointer.StringDeref(resp.ClusterGatewayAddress, "")
	if addr == "" {
		return "", fmt.Errorf("empty cluster gateway address")
	}
	return addr, nil
}