package api

import (
	"context"
	"testing"

	"github.com/golang/mock/gomock"
	"github.com/stretchr/testify/assert"
	cspb "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/service-proxy/pkg/checker/core"

	"github.com/tidbcloud/ticdc-service-ng/pkg/clusterproxy/client/mocks"
)

const (
	testURI = "test-uri"
	taskID  = "test-task-id"
)

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

	mockClient := mocks.NewMockClient(ctrl)
	changefeedAPI := NewChangefeedAPIV2(mockClient)

	ctx := context.Background()
	req := &apiv1.ChangefeedConfig{}

	tidbGroup := &cspb.TiDBGroup{
		Id:           "456",
		SharedPoolId: "1234",
	}
	mockClient.EXPECT().GetTiDBGroup().Return(tidbGroup).AnyTimes()
	mockClient.EXPECT().Request(gomock.Any(), "POST",
		"/proxy/cluster/1234/ticdc/cr/api/v2/changefeeds?keyspace=keyspace", req, gomock.Any()).DoAndReturn(
		func(_ context.Context, method string, relative string, body interface{}, dest interface{}) error {
			dest.(*apiv1.Changefeed).SinkUri = testURI
			return nil
		}).Times(1)
	resp, err := changefeedAPI.CreateChangefeedV2(ctx, "cr", "keyspace", req)
	assert.NoError(t, err)
	assert.Equal(t, testURI, resp.SinkUri)
}

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

	mockClient := mocks.NewMockClient(ctrl)
	tidbGroup := &cspb.TiDBGroup{
		Id:           "456",
		SharedPoolId: "1234",
	}
	mockClient.EXPECT().GetTiDBGroup().Return(tidbGroup).AnyTimes()

	changefeedAPI := NewChangefeedAPIV2(mockClient)

	ctx := context.Background()
	req := &apiv1.ChangefeedConfig{}

	mockClient.EXPECT().Request(gomock.Any(), "PUT",
		"/proxy/cluster/1234/ticdc/cr/api/v2/changefeeds/test-task-id?keyspace=keyspace",
		req, gomock.Any()).DoAndReturn(
		func(ctx context.Context, method string, relative string, body interface{}, dest interface{}) error {
			dest.(*apiv1.Changefeed).SinkUri = testURI
			return nil
		},
	).Times(1)

	resp, err := changefeedAPI.UpdateChangefeedV2(ctx, "cr", "keyspace", taskID, req)
	assert.NoError(t, err)
	assert.Equal(t, testURI, resp.SinkUri)
}

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

	mockClient := mocks.NewMockClient(ctrl)
	tidbGroup := &cspb.TiDBGroup{
		Id:           "456",
		SharedPoolId: "1234",
	}
	mockClient.EXPECT().GetTiDBGroup().Return(tidbGroup).AnyTimes()
	changefeedAPI := NewChangefeedAPIV2(mockClient)

	ctx := context.Background()

	mockClient.EXPECT().Request(gomock.Any(), "GET",
		"/proxy/cluster/1234/ticdc/cr/api/v2/changefeeds?keyspace=keyspace&state=all",
		nil, gomock.Any()).DoAndReturn(
		func(ctx context.Context, method string, relative string, body interface{}, dest interface{}) error {
			dest.(*apiv1.ListChangefeedsResponse).Items = []*apiv1.ChangefeedCommonInfo{{Id: taskID}}
			return nil
		},
	).Times(1)

	resp, err := changefeedAPI.GetChangefeedListV2(ctx, "cr", "keyspace")
	assert.NoError(t, err)
	assert.Equal(t, 1, len(resp.Items))
	assert.Equal(t, taskID, resp.Items[0].Id)
}

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

	mockClient := mocks.NewMockClient(ctrl)
	tidbGroup := &cspb.TiDBGroup{
		Id:           "456",
		SharedPoolId: "1234",
	}
	mockClient.EXPECT().GetTiDBGroup().Return(tidbGroup).AnyTimes()
	changefeedAPI := NewChangefeedAPIV2(mockClient)

	ctx := context.Background()

	mockClient.EXPECT().Request(gomock.Any(), "POST",
		"/proxy/cluster/1234/ticdc/cr/api/v2/changefeeds/test-task-id/pause?keyspace=keyspace",
		nil, nil).Return(nil).Times(1)

	err := changefeedAPI.PauseChangefeedV2(ctx, "cr", "keyspace", taskID)
	assert.NoError(t, err)
}

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

	mockClient := mocks.NewMockClient(ctrl)
	tidbGroup := &cspb.TiDBGroup{
		Id:           "456",
		SharedPoolId: "1234",
	}
	mockClient.EXPECT().GetTiDBGroup().Return(tidbGroup).AnyTimes()
	changefeedAPI := NewChangefeedAPIV2(mockClient)

	ctx := context.Background()
	req := &apiv1.ResumeChangefeedConfig{}

	mockClient.EXPECT().Request(gomock.Any(), "POST",
		"/proxy/cluster/1234/ticdc/cr/api/v2/changefeeds/test-task-id/resume?keyspace=keyspace",
		req, nil).Return(nil).Times(1)

	err := changefeedAPI.ResumeChangefeedV2(ctx, "cr", "keyspace", taskID, req)
	assert.NoError(t, err)
}

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

	mockClient := mocks.NewMockClient(ctrl)
	tidbGroup := &cspb.TiDBGroup{
		Id:           "456",
		SharedPoolId: "1234",
	}
	mockClient.EXPECT().GetTiDBGroup().Return(tidbGroup).AnyTimes()
	changefeedAPI := NewChangefeedAPIV2(mockClient)

	ctx := context.Background()

	mockClient.EXPECT().Request(gomock.Any(), "DELETE",
		"/proxy/cluster/1234/ticdc/cr/api/v2/changefeeds/test-task-id?keyspace=keyspace",
		nil, nil).Return(nil).Times(1)

	err := changefeedAPI.DeleteChangefeedV2(ctx, "cr", "keyspace", taskID)
	assert.NoError(t, err)
}

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

	mockClient := mocks.NewMockClient(ctrl)
	tidbGroup := &cspb.TiDBGroup{
		Id:           "456",
		SharedPoolId: "1234",
	}
	mockClient.EXPECT().GetTiDBGroup().Return(tidbGroup).AnyTimes()
	changefeedAPI := NewChangefeedAPIV2(mockClient)

	ctx := context.Background()

	mockClient.EXPECT().Request(gomock.Any(), "GET",
		"/proxy/cluster/1234/ticdc/cr/api/v2/changefeeds/test-task-id?keyspace=keyspace",
		nil, gomock.Any()).DoAndReturn(
		func(ctx context.Context, method string, relative string, body interface{}, dest interface{}) error {
			dest.(*apiv1.Changefeed).SinkUri = testURI
			return nil
		},
	).Times(1)

	resp, err := changefeedAPI.GetChangefeedV2(ctx, "cr", "keyspace", taskID)
	assert.NoError(t, err)
	assert.Equal(t, testURI, resp.SinkUri)
}

func TestTestConnection(t *testing.T) {
	tests := []struct {
		name      string
		req       *core.CheckRequest
		setupMock func(*mocks.MockClient)
		wantErr   bool
	}{
		{
			name: "success with kafka",
			req: &core.CheckRequest{
				TargetType: "kafka",
				Kafka: &core.KafkaParams{
					Brokers: "localhost:9092",
				},
			},
			setupMock: func(mockClient *mocks.MockClient) {
				mockClient.EXPECT().Request(gomock.Any(), "POST",
					"/check", gomock.Any(), nil).Return(nil).Times(1)
			},
			wantErr: false,
		},
		{
			name:      "nil request",
			req:       nil,
			setupMock: nil,
			wantErr:   true,
		},
		{
			name: "empty target type",
			req: &core.CheckRequest{
				TargetType: "",
				Kafka: &core.KafkaParams{
					Brokers: "localhost:9092",
				},
			},
			setupMock: nil,
			wantErr:   true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()

			mockClient := mocks.NewMockClient(ctrl)
			tidbGroup := &cspb.TiDBGroup{
				Id:           "456",
				SharedPoolId: "1234",
			}
			mockClient.EXPECT().GetTiDBGroup().Return(tidbGroup).AnyTimes()

			if tt.setupMock != nil {
				tt.setupMock(mockClient)
			}

			changefeedAPI := NewChangefeedAPIV2(mockClient)
			ctx := context.Background()
			err := changefeedAPI.TestConnection(ctx, tt.req)
			if tt.wantErr {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}