package client

import (
	"context"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"

	"github.com/golang/mock/gomock"
	"github.com/stretchr/testify/assert"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
)

const (
	proxyEndpoint = "http://proxy.url/cluster/12345/proxy"
)

func TestNewClusterClient(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	ctx := context.Background()
	clusterID := uint64(12345)
	timeout := 10

	opts := []Option{
		WithTimeout(timeout),
	}
	client, err := CreateClient(ctx, clusterID, proxyEndpoint, opts...)
	assert.NoError(t, err)
	assert.NotNil(t, client)
	assert.Equal(t, clusterID, client.GetClusterID())
	assert.Equal(t, proxyEndpoint, client.(*clientImpl).baseURL)
}

func TestNewClusterClient_LocalProxy(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	ctx := context.Background()
	clusterID := uint64(12345)

	opts := []Option{
		WithEndpoint("localhost:8080"),
	}
	client, err := CreateClient(ctx, clusterID, proxyEndpoint, opts...)
	assert.NoError(t, err)
	assert.NotNil(t, client)
	assert.Equal(t, clusterID, client.GetClusterID())
	assert.Equal(t, defaultTimeout, client.(*clientImpl).timeout)
	assert.Equal(t, "http://localhost:8080/cluster/12345/proxy", client.(*clientImpl).baseURL)
}

func TestClientJSONMarshaler(t *testing.T) {
	req := &apiv1.ChangefeedConfig{
		ChangefeedId: "changefeed-1",
		StartTs:      100,
		ReplicaConfig: &apiv1.ReplicaConfig{
			Sink: &apiv1.ReplicaConfig_SinkConfig{
				PulsarConfig: &apiv1.ReplicaConfig_PulsarConfig{
					AuthenticationToken: "token",
					ConnectionTimeout:   5,
				},
			},
		},
	}
	expectedBody := `{"changefeed_id":"changefeed-1","start_ts":100,"replica_config":{"sink":{"pulsar_config":{"authentication-token":"token","connection-timeout":5}}}}`
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(r.Body)
		assert.NoError(t, err)
		assert.Equal(t, expectedBody, strings.ReplaceAll(string(body), " ", ""))
		w.WriteHeader(http.StatusOK)
		w.Write([]byte("{}"))
	}))
	defer ts.Close()

	clusterID := uint64(12345)
	cli, err := CreateClient(context.Background(), clusterID, ts.URL)
	assert.NoError(t, err)

	err = cli.Request(context.Background(), "POST", "/ticdc/changefeed", req, nil)
	assert.NoError(t, err)
}

func TestClientJSONUnmarshaler(t *testing.T) {
	expectedRet := &apiv1.Changefeed{
		Id: "changefeed-1",
		Config: &apiv1.ReplicaConfig{
			Sink: &apiv1.ReplicaConfig_SinkConfig{
				PulsarConfig: &apiv1.ReplicaConfig_PulsarConfig{
					AuthenticationToken: "token",
					ConnectionTimeout:   5,
				},
			},
		},
	}
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		mockResp := `{"upstream_id":123,"id":"changefeed-1","config":{"sink":{"pulsar_config":{"authentication-token":"token","connection-timeout":5}}}}`
		w.WriteHeader(http.StatusOK)
		w.Write([]byte(mockResp))
	}))
	defer ts.Close()

	clusterID := uint64(12345)
	cli, err := CreateClient(context.Background(), clusterID, ts.URL)
	assert.NoError(t, err)

	ret := &apiv1.Changefeed{}
	err = cli.Request(context.Background(), "GET", "/ticdc/changefeed", nil, ret)
	assert.NoError(t, err)
	assert.Equal(t, expectedRet.Id, ret.Id)
	assert.Equal(t, expectedRet.Config.Sink.PulsarConfig.AuthenticationToken, ret.Config.Sink.PulsarConfig.AuthenticationToken)
	assert.Equal(t, expectedRet.Config.Sink.PulsarConfig.ConnectionTimeout, ret.Config.Sink.PulsarConfig.ConnectionTimeout)
}