package impl

import (
	"context"
	"sync"
	"testing"

	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api"
	regionalapi_mock "github.com/tidbcloud/dataflow-service/pkg/regional-api/mock"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
)

type changefeedWorkerRegionalImplTestSuite struct {
	suite.Suite

	mockChangefeedCli    *regionalapi_mock.ChangefeedClient
	mockTiCDCClusterRepo *repo_mock.TiCDCClusterRepo
	mockChangefeedRepo   *repo_mock.TiCDCChangefeedRepo

	worker *changefeedWorkerRegionalImpl
}

func (s *changefeedWorkerRegionalImplTestSuite) SetupSuite() {
	s.mockChangefeedCli = regionalapi_mock.NewChangefeedClient(s.T())
	s.mockTiCDCClusterRepo = repo_mock.NewTiCDCClusterRepo(s.T())
	s.mockChangefeedRepo = repo_mock.NewTiCDCChangefeedRepo(s.T())
	s.worker = &changefeedWorkerRegionalImpl{
		changefeedCli:    s.mockChangefeedCli,
		ticdcClusterRepo: s.mockTiCDCClusterRepo,
		changefeedRepo:   s.mockChangefeedRepo,
	}
}

func (s *changefeedWorkerRegionalImplTestSuite) TestListChangefeedsGroupByTiCDCCluster() {
	s.Run("list ticdc cluster error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockTiCDCClusterRepo.On("List", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		_, err := s.worker.listChangefeedsGroupByTiCDCCluster(context.Background())
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "list ticdc clusters")
	})

	mockTiCDCClusters := []entity.TiCDCCluster{
		{
			Updatable: entity.Updatable{
				Base: entity.Base{
					ID: 1111,
				},
			},
		},
		{
			Updatable: entity.Updatable{
				Base: entity.Base{
					ID: 1112,
				},
			},
		},
	}
	ticdcClusterCall := s.mockTiCDCClusterRepo.
		On("List", mock.Anything, mock.Anything).
		Return(mockTiCDCClusters, nil)
	defer ticdcClusterCall.Unset()

	s.Run("list ticdc cluster error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockChangefeedRepo.On("List", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		_, err := s.worker.listChangefeedsGroupByTiCDCCluster(context.Background())
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "list changefeeds")
	})

	mockChangefeeds := []entity.TiCDCChangefeed{
		{
			Updatable: entity.Updatable{
				Base: entity.Base{
					ID: 11111,
				},
			},
			TiCDCClusterID: 1111,
		},
		{
			Updatable: entity.Updatable{
				Base: entity.Base{
					ID: 11112,
				},
			},
			TiCDCClusterID: 1111,
		},
		{
			Updatable: entity.Updatable{
				Base: entity.Base{
					ID: 11113,
				},
			},
			TiCDCClusterID: 1112,
		},
		{
			Updatable: entity.Updatable{
				Base: entity.Base{
					ID: 11114,
				},
			},
			TiCDCClusterID: 1113,
		},
	}
	changefeedCall := s.mockChangefeedRepo.
		On("List", mock.Anything, mock.Anything).
		Return(mockChangefeeds, nil)
	defer changefeedCall.Unset()

	s.Run("list changefeeds successfully", func() {
		clusterChangefeeds, err := s.worker.listChangefeedsGroupByTiCDCCluster(context.Background())
		s.NoError(err)
		s.Equal(2, len(clusterChangefeeds))
		for i := range clusterChangefeeds {
			if clusterChangefeeds[i].cluster.ID.Uint64() == 1111 {
				s.Equal(2, len(clusterChangefeeds[i].changefeeds))
			} else if clusterChangefeeds[i].cluster.ID.Uint64() == 1112 {
				s.Equal(1, len(clusterChangefeeds[i].changefeeds))
			} else {
				s.Fail("ticdc cluster is is unexpected",
					"ticdc cluster id", clusterChangefeeds[i].cluster.ID)
			}
		}
	})
}

func (s *changefeedWorkerRegionalImplTestSuite) TestSyncOneTiCDCClusterChangefeeds() {
	var wg sync.WaitGroup
	mockClusterChangefeeds := &ticdcClusterChangefeeds{
		cluster: &entity.TiCDCCluster{
			TenantID:          1,
			Region:            "us-west-2",
			Name:              "ticdc-1111",
			InfraProviderType: "eks",
		},
		changefeeds: []*entity.TiCDCChangefeed{
			{
				KernelChangefeedID: "changefeed-1111",
			},
			{
				KernelChangefeedID: "changefeed-1112",
			},
			{
				KernelChangefeedID: "changefeed-1113",
			},
		},
	}
	s.Run("list changefeeds error", func() {
		wg.Add(1)
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockChangefeedCli.
			On("ListChangefeeds", mock.Anything, mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		s.worker.syncOneTiCDCClusterChangefeeds(context.Background(), &wg, mockClusterChangefeeds)
	})

	mockKernelChangefeeds := []*ticdc_v1.ChangefeedCommonInfo{
		{
			Id:            "changefeed-1111",
			CheckpointTso: 123,
		},
		{
			Id:            "changefeed-1112",
			CheckpointTso: 123,
		},
		{
			Id:            "changefeed-1114",
			CheckpointTso: 123,
		},
	}
	listChangefeedsCall := s.mockChangefeedCli.
		On("ListChangefeeds", mock.Anything,
			&regionalapi.Region{
				Name:          "us-west-2",
				TenantID:      1,
				CloudProvider: "aws",
			}, &ticdc_v1.ListChangefeedsRequest{
				TicdcClusterId: "ticdc-1111",
			}).
		Return(&ticdc_v1.ListChangefeedsResponse{
			Items: mockKernelChangefeeds,
		}, nil)
	defer listChangefeedsCall.Unset()
	saveCall := s.mockChangefeedRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Twice()
	defer saveCall.Unset()
	s.Run("sync changefeeds successfully", func() {
		wg.Add(1)
		s.worker.syncOneTiCDCClusterChangefeeds(context.Background(), &wg, mockClusterChangefeeds)
		s.mockChangefeedRepo.AssertExpectations(s.T())
	})
}

func (s *changefeedWorkerRegionalImplTestSuite) TestSyncOneChangefeedStatus() {
	mockChangefeedEnt := &entity.TiCDCChangefeed{}
	mockKernelChangefeed := &ticdc_v1.ChangefeedCommonInfo{
		Id: "changefeed-1111",
	}

	s.Run("save changefeed entity error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockChangefeedRepo.On("Save", mock.Anything, mock.Anything).Return(mockErr)
		defer call.Unset()
		mockKernelChangefeed.CheckpointTso = 12345
		err := s.worker.syncOneChangefeedStatus(context.Background(), mockChangefeedEnt, mockKernelChangefeed)
		s.Error(err)
		s.Equal(mockErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "save changefeed entity")
	})

	call := s.mockChangefeedRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
	defer call.Unset()

	s.Run("sync normal status successfully", func() {
		mockKernelChangefeed.CheckpointTso = 12345
		mockKernelChangefeed.State = "normal"
		err := s.worker.syncOneChangefeedStatus(context.Background(), mockChangefeedEnt, mockKernelChangefeed)
		s.NoError(err)

		matcher := mock.MatchedBy(func(ent *entity.TiCDCChangefeed) bool {
			kernelMetric := ent.KernelMetrics
			return kernelMetric.ID == mockKernelChangefeed.Id && kernelMetric.CheckpointTSO == 12345 &&
				kernelMetric.FeedState == entity.ChangefeedKernelStatusNormal &&
				kernelMetric.UpdatedAt != nil && kernelMetric.RunningError == nil
		})
		s.mockChangefeedRepo.AssertCalled(s.T(), "Save", mock.Anything, matcher)
	})

	s.Run("sync error status successfully", func() {
		mockKernelChangefeed.State = "error"
		mockKernelChangefeed.Error = &ticdc_v1.RunningError{
			Addr:    "address",
			Message: "abc",
			Code:    "123",
		}
		err := s.worker.syncOneChangefeedStatus(context.Background(), mockChangefeedEnt, mockKernelChangefeed)
		s.NoError(err)
		matcher := mock.MatchedBy(func(ent *entity.TiCDCChangefeed) bool {
			kernelMetric := ent.KernelMetrics
			return kernelMetric.FeedState == entity.ChangefeedKernelStatusError &&
				kernelMetric.UpdatedAt != nil && kernelMetric.RunningError != nil &&
				kernelMetric.RunningError.Code == mockKernelChangefeed.Error.Code &&
				kernelMetric.RunningError.Message == mockKernelChangefeed.Error.Message
		})
		s.mockChangefeedRepo.AssertCalled(s.T(), "Save", mock.Anything, matcher)
	})
}

func TestChangefeedWorkerRegionalImplTestSuite(t *testing.T) {
	suite.Run(t, new(changefeedWorkerRegionalImplTestSuite))
}