package service

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"
	regionalpb "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"

	cloudv1 "github.com/tidbcloud/ticdc-service-ng/api/v1"
	cfmocks "github.com/tidbcloud/ticdc-service-ng/pkg/clusterproxy/api/mocks"
	proxymocks "github.com/tidbcloud/ticdc-service-ng/pkg/clusterproxy/mocks"
	"github.com/tidbcloud/ticdc-service-ng/pkg/config"
	regionalservicemocks "github.com/tidbcloud/ticdc-service-ng/pkg/regionalservice/mocks"
)

func TestCreateTiCDCCluster(t *testing.T) {
	scheme := runtime.NewScheme()
	_ = cloudv1.AddToScheme(scheme)

	const (
		testSharedPoolID  = "test-shared-pool"
		testClusterID     = "test-cluster"
		testDataplaneName = "test-dataplane"
	)

	fullReq := &apiv1.CreateTiCDCClusterRequest{
		TicdcClusterId: testClusterID,
		TicdcCluster: &apiv1.TiCDCCluster{
			SharedPoolId:  testSharedPoolID,
			NodeSpec:      &apiv1.NodeInfo{Count: 1, CpuCore: 2, MemoryGi: 4},
			Storage:       &apiv1.TiCDCCluster_Storage{Type: "gp3", SizeGi: 100},
			Configuration: &apiv1.TiCDCCluster_Configuration{Version: "v6.5.0"},
		},
	}

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	t.Run("Successful creation", func(t *testing.T) {
		mockRegionalCli := regionalservicemocks.NewMockClient(ctrl)
		mockRegionalCli.EXPECT().GetSharedPoolByID(gomock.Any(), testSharedPoolID).Return(&cspb.SharedPool{DataplaneName: testDataplaneName}, nil)
		mockRegionalCli.EXPECT().GetDataPlaneByID(gomock.Any(), testDataplaneName).Return(&regionalpb.DataPlane{Kubernetes: &regionalpb.DataPlane_Kubernetes{Version: "1.21"}}, nil)

		server := cdcServer{
			kubeClient:  fake.NewClientBuilder().WithScheme(scheme).Build(),
			cfg:         &config.Config{Namespace: "default"},
			regionalCli: mockRegionalCli,
		}

		resp, err := server.CreateTiCDCCluster(context.Background(), fullReq)

		assert.NoError(t, err)
		assert.NotNil(t, resp)
		assert.Equal(t, testClusterID, resp.Id)
	})

	t.Run("SharedPool not found", func(t *testing.T) {
		mockRegionalCli := regionalservicemocks.NewMockClient(ctrl)
		mockRegionalCli.EXPECT().GetSharedPoolByID(gomock.Any(), testSharedPoolID).Return(nil, apierrors.NewNotFound(schema.GroupResource{}, testSharedPoolID))

		server := cdcServer{
			kubeClient:  fake.NewClientBuilder().WithScheme(scheme).Build(),
			cfg:         &config.Config{Namespace: "default"},
			regionalCli: mockRegionalCli,
		}

		_, err := server.CreateTiCDCCluster(context.Background(), fullReq)

		assert.Error(t, err)
		st, ok := status.FromError(err)
		assert.True(t, ok)
		assert.Equal(t, codes.NotFound, st.Code())
	})

	t.Run("Dataplane not found", func(t *testing.T) {
		mockRegionalCli := regionalservicemocks.NewMockClient(ctrl)
		mockRegionalCli.EXPECT().GetSharedPoolByID(gomock.Any(), testSharedPoolID).Return(&cspb.SharedPool{DataplaneName: testDataplaneName}, nil)
		mockRegionalCli.EXPECT().GetDataPlaneByID(gomock.Any(), testDataplaneName).Return(nil, apierrors.NewNotFound(schema.GroupResource{}, testDataplaneName))

		server := cdcServer{
			kubeClient:  fake.NewClientBuilder().WithScheme(scheme).Build(),
			cfg:         &config.Config{Namespace: "default"},
			regionalCli: mockRegionalCli,
		}

		_, err := server.CreateTiCDCCluster(context.Background(), fullReq)

		assert.Error(t, err)
		st, ok := status.FromError(err)
		assert.True(t, ok)
		assert.Equal(t, codes.Internal, st.Code())
	})

	t.Run("Cluster already exists", func(t *testing.T) {
		mockRegionalCli := regionalservicemocks.NewMockClient(ctrl)
		mockRegionalCli.EXPECT().GetSharedPoolByID(gomock.Any(), testSharedPoolID).Return(&cspb.SharedPool{DataplaneName: testDataplaneName}, nil)
		mockRegionalCli.EXPECT().GetDataPlaneByID(gomock.Any(), testDataplaneName).Return(&regionalpb.DataPlane{Kubernetes: &regionalpb.DataPlane_Kubernetes{Version: "1.21"}}, nil)

		mockKubeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&cloudv1.TiCDC{
			ObjectMeta: metav1.ObjectMeta{Name: testClusterID, Namespace: "default"},
		}).Build()

		server := cdcServer{
			kubeClient:  mockKubeClient,
			cfg:         &config.Config{Namespace: "default"},
			regionalCli: mockRegionalCli,
		}

		_, err := server.CreateTiCDCCluster(context.Background(), fullReq)

		assert.Error(t, err)
		st, ok := status.FromError(err)
		assert.True(t, ok)
		assert.Equal(t, codes.AlreadyExists, st.Code())
	})
}

func TestListTiCDCClusters(t *testing.T) {
	scheme := runtime.NewScheme()
	_ = cloudv1.AddToScheme(scheme)

	clusters := []client.Object{
		&cloudv1.TiCDC{
			ObjectMeta: metav1.ObjectMeta{Name: "cluster1", Namespace: "default"},
			Spec:       cloudv1.TiCDCSpec{Replicas: 1},
		},
		&cloudv1.TiCDC{
			ObjectMeta: metav1.ObjectMeta{Name: "cluster2", Namespace: "default"},
			Spec:       cloudv1.TiCDCSpec{Replicas: 2},
		},
	}

	mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(clusters...).Build()

	server := cdcServer{
		kubeClient: mockClient,
		cfg:        &config.Config{Namespace: "default"},
	}

	req := &apiv1.ListTiCDCClustersRequest{}
	resp, err := server.ListTiCDCClusters(context.Background(), req)

	assert.NoError(t, err)
	assert.Len(t, resp.TicdcClusters, 2)
}

func TestGetTiCDCCluster(t *testing.T) {
	scheme := runtime.NewScheme()
	_ = cloudv1.AddToScheme(scheme)

	const testClusterID = "test-cluster"

	t.Run("Successful get", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&cloudv1.TiCDC{
			ObjectMeta: metav1.ObjectMeta{Name: testClusterID, Namespace: "default"},
			Spec: cloudv1.TiCDCSpec{
				Replicas: 1,
				NodeSpec: cloudv1.NodeSpec{
					CpuCore:  1,
					MemoryGi: 8,
				},
				Storage: cloudv1.StorageSpec{
					Type:          "gp3",
					SizeGi:        100,
					Iops:          1000,
					ThroughputMbs: 1000,
				},
			},
		}).Build()
		server := cdcServer{kubeClient: mockClient, cfg: &config.Config{Namespace: "default"}}
		req := &apiv1.GetTiCDCClusterRequest{TicdcClusterId: testClusterID}
		resp, err := server.GetTiCDCCluster(context.Background(), req)
		assert.NoError(t, err)
		assert.NotNil(t, resp)
		assert.Equal(t, testClusterID, resp.Id)
	})

	t.Run("Cluster not found", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).Build()
		server := cdcServer{kubeClient: mockClient, cfg: &config.Config{Namespace: "default"}}
		req := &apiv1.GetTiCDCClusterRequest{TicdcClusterId: "non-existent"}
		_, err := server.GetTiCDCCluster(context.Background(), req)
		assert.Error(t, err)
		st, _ := status.FromError(err)
		assert.Equal(t, codes.NotFound, st.Code())
	})
}

func TestScaleTiCDCCluster(t *testing.T) {
	scheme := runtime.NewScheme()
	_ = cloudv1.AddToScheme(scheme)
	const testClusterID = "test-cluster"

	t.Run("Successful scale", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&cloudv1.TiCDC{
			ObjectMeta: metav1.ObjectMeta{Name: testClusterID, Namespace: "default"},
			Spec:       cloudv1.TiCDCSpec{Replicas: 1},
		}).Build()
		server := cdcServer{kubeClient: mockClient, cfg: &config.Config{Namespace: "default"}}
		req := &apiv1.ScaleTiCDCClusterRequest{
			TicdcClusterId: testClusterID,
			NodeSpec: &apiv1.NodeInfo{
				Count: 3,
			},
			Storage: &apiv1.TiCDCCluster_Storage{
				Type:          "gp3",
				SizeGi:        100,
				Iops:          1000,
				ThroughputMbs: 100,
			},
		}
		_, err := server.ScaleTiCDCCluster(context.Background(), req)
		assert.NoError(t, err)
		updated := &cloudv1.TiCDC{}
		_ = mockClient.Get(context.Background(), client.ObjectKey{Name: testClusterID, Namespace: "default"}, updated)
		assert.Equal(t, 3, updated.Spec.Replicas)
		assert.Equal(t, uint32(1000), updated.Spec.Storage.Iops)
		assert.Equal(t, uint32(100), updated.Spec.Storage.ThroughputMbs)
		assert.Equal(t, uint32(100), updated.Spec.Storage.SizeGi)
		assert.Equal(t, "gp3", updated.Spec.Storage.Type)
	})
}

func TestUpdateTiCDCClusterConfig(t *testing.T) {
	scheme := runtime.NewScheme()
	_ = cloudv1.AddToScheme(scheme)
	const testClusterID = "test-cluster"

	t.Run("Successful update", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&cloudv1.TiCDC{
			ObjectMeta: metav1.ObjectMeta{Name: testClusterID, Namespace: "default"},
			Spec:       cloudv1.TiCDCSpec{Configuration: &cloudv1.TiCDCConfiguration{Version: "v1"}},
		}).Build()
		server := cdcServer{kubeClient: mockClient, cfg: &config.Config{Namespace: "default"}}
		req := &apiv1.UpdateTiCDCClusterConfigRequest{
			TicdcClusterId: testClusterID,
			Configuration:  &apiv1.TiCDCCluster_Configuration{Version: "v2"},
		}
		_, err := server.UpdateTiCDCClusterConfig(context.Background(), req)
		assert.NoError(t, err)
		updated := &cloudv1.TiCDC{}
		_ = mockClient.Get(context.Background(), client.ObjectKey{Name: testClusterID, Namespace: "default"}, updated)
		assert.Equal(t, "v2", updated.Spec.Configuration.Version)
	})
}

func TestUpdateTiCDCClusterStorage(t *testing.T) {
	scheme := runtime.NewScheme()
	_ = cloudv1.AddToScheme(scheme)
	const testClusterID = "test-cluster"

	t.Run("Successful update", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&cloudv1.TiCDC{
			ObjectMeta: metav1.ObjectMeta{Name: testClusterID, Namespace: "default"},
			Spec:       cloudv1.TiCDCSpec{Storage: cloudv1.StorageSpec{SizeGi: 100}},
		}).Build()
		server := cdcServer{kubeClient: mockClient, cfg: &config.Config{Namespace: "default"}}
		req := &apiv1.UpdateTiCDCClusterStorageRequest{
			TicdcClusterId: testClusterID,
			Storage:        &apiv1.TiCDCCluster_Storage{SizeGi: 200},
		}
		_, err := server.UpdateTiCDCClusterStorage(context.Background(), req)
		assert.NoError(t, err)
		updated := &cloudv1.TiCDC{}
		_ = mockClient.Get(context.Background(), client.ObjectKey{Name: testClusterID, Namespace: "default"}, updated)
		assert.Equal(t, uint32(200), updated.Spec.Storage.SizeGi)
	})
}

func TestDeleteTiCDCCluster(t *testing.T) {
	scheme := runtime.NewScheme()
	_ = cloudv1.AddToScheme(scheme)
	const testClusterID = "test-cluster"

	t.Run("Successful delete", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&cloudv1.TiCDC{
			ObjectMeta: metav1.ObjectMeta{Name: testClusterID, Namespace: "default"},
		}).Build()
		server := cdcServer{kubeClient: mockClient, cfg: &config.Config{Namespace: "default"}}
		req := &apiv1.DeleteTiCDCClusterRequest{TicdcClusterId: testClusterID}
		_, err := server.DeleteTiCDCCluster(context.Background(), req)
		assert.NoError(t, err)
		err = mockClient.Get(context.Background(), client.ObjectKey{Name: testClusterID, Namespace: "default"}, &cloudv1.TiCDC{})
		assert.True(t, apierrors.IsNotFound(err))
	})
}

func TestRestartTiCDCCluster(t *testing.T) {
	scheme := runtime.NewScheme()
	_ = cloudv1.AddToScheme(scheme)
	const testClusterID = "test-cluster"

	t.Run("Successful restart", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&cloudv1.TiCDC{
			ObjectMeta: metav1.ObjectMeta{Name: testClusterID, Namespace: "default"},
		}).Build()
		server := cdcServer{kubeClient: mockClient, cfg: &config.Config{Namespace: "default"}}
		req := &apiv1.RestartTiCDCClusterRequest{TicdcClusterId: testClusterID}
		_, err := server.RestartTiCDCCluster(context.Background(), req)
		assert.NoError(t, err)
		updated := &cloudv1.TiCDC{}
		_ = mockClient.Get(context.Background(), client.ObjectKey{Name: testClusterID, Namespace: "default"}, updated)
		assert.Contains(t, updated.Annotations, "ticdc.cloud.pingcap.com/restart")
	})
}

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

	const (
		ticdcClusterID = "ticdc-cluster-1"
		tidbClusterID  = "tidb-cluster-1"
		changefeedID   = "cf-1"
	)

	mockChangefeedClient := cfmocks.NewMockChangefeedAPIV2(ctrl)
	mockProxy := proxymocks.NewMockClusterProxy(ctrl)
	mockProxy.EXPECT().ChangefeedClient().Return(mockChangefeedClient).AnyTimes()

	mockProxyModel := proxymocks.NewMockModel(ctrl)
	mockProxyModel.EXPECT().
		GetClusterProxy(gomock.Any(), gomock.Any(), gomock.Any()).Return(mockProxy, nil).
		AnyTimes()

	server := cdcServer{
		proxyModel: mockProxyModel,
	}

	t.Run("CreateChangefeed", func(t *testing.T) {
		req := &apiv1.CreateChangefeedRequest{
			TicdcClusterId: ticdcClusterID,
			TidbClusterId:  tidbClusterID,
			ChangefeedId:   changefeedID,
			Changefeed:     &apiv1.ChangefeedConfig{},
		}
		mockChangefeedClient.EXPECT().CreateChangefeedV2(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(&apiv1.Changefeed{Id: changefeedID}, nil)
		resp, err := server.CreateChangefeed(context.Background(), req)
		assert.NoError(t, err)
		assert.Equal(t, changefeedID, resp.Id)
	})

	t.Run("ListChangefeeds", func(t *testing.T) {
		req := &apiv1.ListChangefeedsRequest{
			TicdcClusterId: ticdcClusterID,
			TidbClusterId:  tidbClusterID,
		}
		mockChangefeedClient.EXPECT().GetChangefeedListV2(gomock.Any(), gomock.Any(), gomock.Any()).
			Return(&apiv1.ListChangefeedsResponse{Items: []*apiv1.ChangefeedCommonInfo{{Id: changefeedID}}}, nil)
		resp, err := server.ListChangefeeds(context.Background(), req)
		assert.NoError(t, err)
		assert.Len(t, resp.Items, 1)
		assert.Equal(t, changefeedID, resp.Items[0].Id)
	})

	t.Run("GetChangefeed", func(t *testing.T) {
		req := &apiv1.GetChangefeedRequest{
			TicdcClusterId: ticdcClusterID,
			TidbClusterId:  tidbClusterID,
			ChangefeedId:   changefeedID,
		}
		mockChangefeedClient.EXPECT().GetChangefeedV2(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(&apiv1.Changefeed{Id: changefeedID}, nil)
		resp, err := server.GetChangefeed(context.Background(), req)
		assert.NoError(t, err)
		assert.Equal(t, changefeedID, resp.Id)
	})

	t.Run("PauseChangefeed", func(t *testing.T) {
		req := &apiv1.PauseChangefeedRequest{
			TicdcClusterId: ticdcClusterID,
			TidbClusterId:  tidbClusterID,
			ChangefeedId:   changefeedID,
		}
		mockChangefeedClient.EXPECT().PauseChangefeedV2(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
		_, err := server.PauseChangefeed(context.Background(), req)
		assert.NoError(t, err)
	})

	t.Run("ResumeChangefeed", func(t *testing.T) {
		req := &apiv1.ResumeChangefeedRequest{
			TicdcClusterId: ticdcClusterID,
			TidbClusterId:  tidbClusterID,
			ChangefeedId:   changefeedID,
		}
		mockChangefeedClient.EXPECT().ResumeChangefeedV2(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
		_, err := server.ResumeChangefeed(context.Background(), req)
		assert.NoError(t, err)
	})

	t.Run("UpdateChangefeed", func(t *testing.T) {
		req := &apiv1.UpdateChangefeedRequest{
			TicdcClusterId: ticdcClusterID,
			TidbClusterId:  tidbClusterID,
			ChangefeedId:   changefeedID,
			Changefeed:     &apiv1.ChangefeedConfig{},
		}
		mockChangefeedClient.EXPECT().UpdateChangefeedV2(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(&apiv1.Changefeed{Id: changefeedID}, nil)
		resp, err := server.UpdateChangefeed(context.Background(), req)
		assert.NoError(t, err)
		assert.Equal(t, changefeedID, resp.Id)
	})

	t.Run("DeleteChangefeed", func(t *testing.T) {
		req := &apiv1.DeleteChangefeedRequest{
			TicdcClusterId: ticdcClusterID,
			TidbClusterId:  tidbClusterID,
			ChangefeedId:   changefeedID,
		}
		mockChangefeedClient.EXPECT().DeleteChangefeedV2(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
		_, err := server.DeleteChangefeed(context.Background(), req)
		assert.NoError(t, err)
	})
}