package impl

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"testing"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	internal_config "github.com/tidbcloud/dataflow-service/internal/config"
	encryptor_mock "github.com/tidbcloud/dataflow-service/pkg/encryptor-wrapper/mock"
	grpcli_mock "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli/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"
	"github.com/tidbcloud/pkgv2/clusterproxy"
)

type serverlessDownstreamAdaptorTestSuite struct {
	suite.Suite
	serverlessAdaptor *serverlessDownstreamAdaptor

	mockPreCheckCli    *regionalapi_mock.TiCDCPreCheckClient
	mockEncryptor      *encryptor_mock.Encryptor
	mockPLEndpointRepo *repo_mock.BizPrivateLinkEndpointRepo
	mockCoreCli        *grpcli_mock.CoreGrpcClient
}

func (s *serverlessDownstreamAdaptorTestSuite) SetupSuite() {
	s.mockPreCheckCli = regionalapi_mock.NewTiCDCPreCheckClient(s.T())
	s.mockEncryptor = encryptor_mock.NewEncryptor(s.T())
	s.mockPLEndpointRepo = repo_mock.NewBizPrivateLinkEndpointRepo(s.T())
	s.mockCoreCli = grpcli_mock.NewCoreGrpcClient(s.T())

	s.serverlessAdaptor = &serverlessDownstreamAdaptor{
		encryptor:               s.mockEncryptor,
		privatelinkEndpointRepo: s.mockPLEndpointRepo,
		coreGrpcClient:          s.mockCoreCli,
		mysqlAdaptor: &mysqlDownstreamAdaptor{
			cdcPreCheckCli: s.mockPreCheckCli,
			encryptor:      s.mockEncryptor,
		},
	}
}

func (s *serverlessDownstreamAdaptorTestSuite) TestParseServerlessParams() {
	testcases := []struct {
		name             string
		downstreamParams interface{}
		expectedError    bool
	}{
		{
			name:             "nil param",
			downstreamParams: nil,
			expectedError:    true,
		},
		{
			name:             "wrong type param",
			downstreamParams: &dataflow.MySQLInfo{},
			expectedError:    true,
		},
		{
			name:             "wrong type param2",
			downstreamParams: dataflow.ServerlessTidbInfo{},
			expectedError:    true,
		},
		{
			name:             "right type param",
			downstreamParams: &dataflow.ServerlessTidbInfo{},
			expectedError:    false,
		},
	}

	for _, tc := range testcases {
		s.Run(tc.name, func() {
			_, err := s.serverlessAdaptor.parseServerlessParams(tc.downstreamParams)
			if tc.expectedError {
				s.Error(err)
			} else {
				s.NoError(err)
			}
		})
	}
}

func (s *serverlessDownstreamAdaptorTestSuite) TestParseServerlessParamsStr() {
	testcases := []struct {
		name             string
		downstreamParams interface{}
		expectedError    bool
	}{
		{
			name: "wrong type param",
			downstreamParams: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "abc:3306",
				},
			},
			expectedError: true,
		},
		{
			name: "right type param",
			downstreamParams: &dataflow.ServerlessTidbInfo{
				ClusterId: 111,
				MysqlInfo: &dataflow.MySQLInfo{
					Connection: &dataflow.Connection{
						Endpoint: "abc:3306",
					},
				},
			},
			expectedError: false,
		},
	}

	for _, tc := range testcases {
		s.Run(tc.name, func() {
			paramsBytes, err := json.Marshal(tc.downstreamParams)
			s.Require().NoError(err)
			_, err = s.serverlessAdaptor.parseServerlessParamsStr(string(paramsBytes))
			if tc.expectedError {
				s.Error(err)
			} else {
				s.NoError(err)
			}
		})
	}
}

func (s *serverlessDownstreamAdaptorTestSuite) TestQueryRegionalDNSByPlEndpointID() {
	mockID := uint64(111)
	s.Run("load pl endpoint error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockPLEndpointRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		_, err := s.serverlessAdaptor.queryRegionalDNSByPlEndpointID(context.Background(), mockID)
		s.Error(err)
		s.Contains(err.Error(), "query private link endpoint")
	})

	mockPLEndpoint := &entity.BizPrivateLinkEndpoint{
		Status: entity.BizPLEndpointStatusCreating,
	}
	call := s.mockPLEndpointRepo.On("Load", mock.Anything, mockID).Return(mockPLEndpoint, nil)
	defer call.Unset()
	s.Run("pl endpoint is not available", func() {
		_, err := s.serverlessAdaptor.queryRegionalDNSByPlEndpointID(context.Background(), mockID)
		s.Error(err)
		s.Contains(err.Error(), "network connection is timed out")
	})

	mockPLEndpoint.Status = entity.BizPLEndpointStatusAvailable
	mockPLEndpoint.EndpointAddress = "region.abc.com"
	s.Run("pl endpoint is available", func() {
		addr, err := s.serverlessAdaptor.queryRegionalDNSByPlEndpointID(context.Background(), mockID)
		s.NoError(err)
		s.Equal(mockPLEndpoint.EndpointAddress, addr)
	})
}

func (s *serverlessDownstreamAdaptorTestSuite) TestQueryServerlessTiDBPort() {
	mockServerlessClusterID := uint64(111)
	s.Run("get serverless cluster error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "get serverless cluster error")
		call := s.mockCoreCli.On("GetAvailableClusterByID", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		_, err := s.serverlessAdaptor.queryServerlessTiDBPort(context.Background(), mockServerlessClusterID)
		s.Error(err)
		s.Contains(err.Error(), "get serverless TiDB cluster")
	})

	mockServerlessCluster := &core.Cluster{
		Id: 111,
	}
	call := s.mockCoreCli.
		On("GetAvailableClusterByID", mock.Anything, mockServerlessClusterID).
		Return(mockServerlessCluster, nil)
	defer call.Unset()
	s.Run("port not found", func() {
		_, err := s.serverlessAdaptor.queryServerlessTiDBPort(context.Background(), mockServerlessClusterID)
		s.Error(err)
		s.Contains(err.Error(), "port of serverless TiDB cluster is not found")
	})

	mockPort := int32(123)
	mockServerlessCluster.Connections = []*core.ClusterConnection{
		{
			CreateSuccess: true,
			Port:          mockPort,
		},
	}
	s.Run("get port successfully", func() {
		port, err := s.serverlessAdaptor.queryServerlessTiDBPort(context.Background(), mockServerlessClusterID)
		s.NoError(err)
		s.Equal(mockPort, port)
	})
}

func (s *serverlessDownstreamAdaptorTestSuite) TestFillServerlessEndpoint() {
	mockServerlessClusterID := uint64(111)
	mockPLEndpointID := uint64(222)
	mockServerlessParams := &dataflow.ServerlessTidbInfo{
		ClusterId: mockServerlessClusterID,
		MysqlInfo: &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{},
			PrivatelinkInfo: &dataflow.PrivatelinkInfo{
				DfPrivatelinkEndpointId: mockPLEndpointID,
			},
		},
	}
	s.Run("query pl endpoint error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockPLEndpointRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		err := s.serverlessAdaptor.fillServerlessEndpoint(context.Background(), mockServerlessParams)
		s.Error(err)
		s.Contains(err.Error(), "query private link endpoint")
	})

	mockPLEndpoint := &entity.BizPrivateLinkEndpoint{
		Status:          entity.BizPLEndpointStatusAvailable,
		EndpointAddress: "region.abc.com",
	}
	plCall := s.mockPLEndpointRepo.On("Load", mock.Anything, mockPLEndpointID).Return(mockPLEndpoint, nil)
	defer plCall.Unset()

	s.Run("query serverless tidb port error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "get serverless cluster error")
		call := s.mockCoreCli.On("GetAvailableClusterByID", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		err := s.serverlessAdaptor.fillServerlessEndpoint(context.Background(), mockServerlessParams)
		s.Error(err)
		s.Contains(err.Error(), "get serverless TiDB cluster")
	})

	mockServerlessCluster := &core.Cluster{
		Id: 111,
		Connections: []*core.ClusterConnection{
			{
				CreateSuccess: true,
				Port:          123,
			},
		},
	}
	clusterCall := s.mockCoreCli.
		On("GetAvailableClusterByID", mock.Anything, mockServerlessClusterID).
		Return(mockServerlessCluster, nil)
	defer clusterCall.Unset()

	s.Run("fill serverless endpoint successfully", func() {
		err := s.serverlessAdaptor.fillServerlessEndpoint(context.Background(), mockServerlessParams)
		s.NoError(err)
		s.Equal("region.abc.com:123", mockServerlessParams.MysqlInfo.Connection.Endpoint)
	})
}

func (s *serverlessDownstreamAdaptorTestSuite) TestTestConnection() {
	cluster := &core.Cluster{
		Id:                111,
		ProjectId:         11,
		TenantId:          1,
		RegionName:        "us-west-2",
		InfraProviderType: "eks",
	}

	s.Run("downstream param error", func() {
		downstreamParams := &dataflow.MySQLInfo{}
		err := s.serverlessAdaptor.TestConnection(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

	mockServerlessParams := &dataflow.ServerlessTidbInfo{
		ClusterId: 111,
		MysqlInfo: &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Username: "b",
				Password: "c",
			},
			PrivatelinkInfo: &dataflow.PrivatelinkInfo{
				DfPrivatelinkEndpointId: 222,
			},
		},
	}

	s.Run("fill serverless endpoint error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockPLEndpointRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		err := s.serverlessAdaptor.TestConnection(context.Background(), cluster, mockServerlessParams)
		s.Error(err)
		s.Contains(err.Error(), "fill serverless endpoint")
	})

	mockPLEndpoint := &entity.BizPrivateLinkEndpoint{
		Status:          entity.BizPLEndpointStatusAvailable,
		EndpointAddress: "region.abc.com",
	}
	plCall := s.mockPLEndpointRepo.On("Load", mock.Anything, mock.Anything).Return(mockPLEndpoint, nil)
	defer plCall.Unset()
	mockServerlessCluster := &core.Cluster{
		Id: 111,
		Connections: []*core.ClusterConnection{
			{
				CreateSuccess: true,
				Port:          123,
			},
		},
	}
	clusterCall := s.mockCoreCli.
		On("GetAvailableClusterByID", mock.Anything, mock.Anything).
		Return(mockServerlessCluster, nil)
	defer clusterCall.Unset()

	s.Run("test connection error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockPreCheckCli.
			On("TestNetworkConnection", mock.Anything, mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()
		err := s.serverlessAdaptor.TestConnection(context.Background(), cluster, mockServerlessParams)
		s.Error(err)
	})

	s.Run("test connection successfully", func() {
		expectedRegion := &regionalapi.Region{
			Name:          "us-west-2",
			TenantID:      1,
			CloudProvider: "aws",
		}
		expectedReq := &ticdc_v1.TestNetworkConnectionRequest{
			ClusterId: "111",
			Type:      "mysql",
			MysqlParam: &ticdc_v1.TestNetworkConnectionRequest_MySQLParameter{
				Url:      "region.abc.com:123",
				User:     "b",
				Password: "c",
			},
		}
		call := s.mockPreCheckCli.
			On("TestNetworkConnection", mock.Anything, expectedRegion, expectedReq).
			Return(nil)
		defer call.Unset()
		err := s.serverlessAdaptor.TestConnection(context.Background(), cluster, mockServerlessParams)
		s.NoError(err)
		s.Equal("region.abc.com:123", mockServerlessParams.MysqlInfo.Connection.Endpoint)
	})
}

func (s *serverlessDownstreamAdaptorTestSuite) TestPreCheckChangefeedParameters() {
	cluster := &core.Cluster{
		Id:                111,
		ProjectId:         11,
		TenantId:          1,
		RegionName:        "us-west-2",
		InfraProviderType: "eks",
	}

	s.Run("downstream param error", func() {
		downstreamParams := &dataflow.MySQLInfo{}
		err := s.serverlessAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

	serverlessParams := &dataflow.ServerlessTidbInfo{
		ClusterId: 111,
		MysqlInfo: nil,
	}
	s.Run("precheck mysql param error", func() {
		err := s.serverlessAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, serverlessParams)
		s.Error(err)
	})

	mockTSO := uint64(123123)
	serverlessParams.MysqlInfo = &dataflow.MySQLInfo{
		StartPosition: &dataflow.StartPosition{
			Mode: dataflow.StartMode_FROM_TSO,
			Tso:  mockTSO,
		},
	}
	s.Run("precheck successfully", func() {
		expectedRegion := &regionalapi.Region{
			Name:          "us-west-2",
			TenantID:      1,
			CloudProvider: "aws",
		}
		expectedReq := &ticdc_v1.ValidateChangefeedParametersRequest{
			ClusterId: "111",
			Type:      "mysql",
			StartTso:  123123,
		}
		call := s.mockPreCheckCli.
			On("ValidateChangefeedParameters", mock.Anything, expectedRegion, expectedReq).
			Return(nil)
		defer call.Unset()

		err := s.serverlessAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, serverlessParams)
		s.Assert().NoError(err)
	})
}

func (s *serverlessDownstreamAdaptorTestSuite) TestEncryptPasswordInPlace() {
	downstreamParams := &dataflow.ServerlessTidbInfo{
		MysqlInfo: &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Password: "abc",
			},
		},
	}
	encryptedPassword := base64.StdEncoding.EncodeToString([]byte(downstreamParams.MysqlInfo.Connection.Password))
	call := s.mockEncryptor.
		On("EncryptBase64Inplace", &downstreamParams.MysqlInfo.Connection.Password).
		Return(nil).
		Run(func(args mock.Arguments) {
			password := args.Get(0).(*string)
			*password = encryptedPassword
		})
	defer call.Unset()

	err := s.serverlessAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
	s.Assert().NoError(err)
	s.Assert().Equal(encryptedPassword, downstreamParams.MysqlInfo.Connection.Password)
}

func (s *serverlessDownstreamAdaptorTestSuite) TestAssembleChangefeedConfig() {
	cluster := &core.Cluster{
		Id:        111,
		ProjectId: 11,
		TenantId:  1,
	}
	downstreamParams := &dataflow.ServerlessTidbInfo{
		MysqlInfo: &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Endpoint: "region.abc.com:123",
				Username: "user",
				Password: "encrypted-password",
			},
		},
	}

	s.Run("assemble uri config error", func() {
		decryptErr := bizerr.New(bizerr.ErrCodeUnknown, "decrypt error")
		call := s.mockEncryptor.
			On("DecryptBase64", mock.Anything).
			Return("", decryptErr)
		defer call.Unset()
		downstreamParamsBytes, err := json.Marshal(downstreamParams)
		s.Require().NoError(err)
		bizCf := &entity.BizChangefeed{
			DownstreamParams: string(downstreamParamsBytes),
		}
		_, err = s.serverlessAdaptor.AssembleChangefeedConfig(context.Background(), cluster, bizCf)
		s.Error(err)
		s.Contains(err.Error(), "assemble uri config")
	})

	decryptedPassword := "raw-password"
	decryptCall := s.mockEncryptor.
		On("DecryptBase64", mock.Anything).
		Return(decryptedPassword, nil)
	defer decryptCall.Unset()
	downstreamParams.MysqlInfo.StartPosition = &dataflow.StartPosition{
		Mode: dataflow.StartMode_FROM_UTC,
		Utc:  time.Now().Format(time.RFC1123),
	}
	s.Run("assemble changefeed config data error", func() {
		downstreamParamsBytes, err := json.Marshal(downstreamParams)
		s.Require().NoError(err)
		bizCf := &entity.BizChangefeed{
			DownstreamParams: string(downstreamParamsBytes),
		}
		_, err = s.serverlessAdaptor.AssembleChangefeedConfig(context.Background(), cluster, bizCf)
		s.Error(err)
		s.Contains(err.Error(), "assemble changefeed config")
	})

	downstreamParams.MysqlInfo.StartPosition = &dataflow.StartPosition{
		Mode: dataflow.StartMode_FROM_TSO,
		Tso:  123123,
	}
	downstreamParams.MysqlInfo.TableConfig = &dataflow.TableConfig{
		Mode:       dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE,
		FilterRule: []string{"*.*"},
		EventFilterRule: []*dataflow.EventFilterRule{
			{
				Matcher:     []string{"test.*"},
				IgnoreEvent: []string{"delete"},
			},
		},
		CaseSensitive: true,
	}
	sqlModeConfig := internal_config.SQLModeConfig{
		SQLMode:   "a",
		TenantID:  []string{"1"},
		ProjectID: []string{"11"},
		ClusterID: []string{"111"},
	}
	internal_config.Global().ChangefeedConfig.SQLModeConfig = []internal_config.SQLModeConfig{sqlModeConfig}

	expectedChangefeedConfig := &domain.TiCDCChangefeedConfig{
		URIConfig: &domain.TiCDCChangefeedMySQLURIConfigData{
			Endpoint:             pointer.ToString("region.abc.com:123"),
			User:                 pointer.ToString("user"),
			Password:             pointer.ToString(decryptedPassword),
			TransactionAtomicity: pointer.ToString("none"),
		},
		Config: &domain.TiCDCChangefeedConfigData{
			StartTs: pointer.ToUint64(123123),
			ReplicaConfig: &clusterproxy.ReplicaConfig{
				CaseSensitive:         pointer.ToBool(true),
				ForceReplicate:        pointer.ToBool(false),
				IgnoreIneligibleTable: pointer.ToBool(true),
				SQLMode:               "a",
				Filter: &clusterproxy.FilterConfig{
					Rules: []string{"*.*"},
					EventFilters: []clusterproxy.EventFilterRule{
						{
							Matcher:     []string{"test.*"},
							IgnoreEvent: []string{"delete"},
						},
					},
				},
			},
		},
	}
	s.Run("assemble changefeed config data successfully", func() {
		downstreamParamsBytes, err := json.Marshal(downstreamParams)
		s.Require().NoError(err)
		bizCf := &entity.BizChangefeed{
			DownstreamParams: string(downstreamParamsBytes),
		}
		changefeedConfig, err := s.serverlessAdaptor.AssembleChangefeedConfig(context.Background(), cluster, bizCf)
		s.NoError(err)
		s.Equal(expectedChangefeedConfig, changefeedConfig)
	})
}

func (s *serverlessDownstreamAdaptorTestSuite) TestCompleteEditParams() {
	currentParams := &dataflow.ServerlessTidbInfo{
		ClusterId: 111,
		MysqlInfo: &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Endpoint: "region.abc.com:123",
				Username: "user",
				Password: "encrypted-password",
			},
			PrivatelinkInfo: &dataflow.PrivatelinkInfo{
				DfPrivatelinkEndpointId: 222,
			},
		},
	}
	currentParamsBytes, err := json.Marshal(currentParams)
	s.Require().NoError(err)
	currentParamsStr := string(currentParamsBytes)

	s.Run("pl endpoint id is changed", func() {
		editParams := &dataflow.ServerlessTidbInfo{
			ClusterId: 111,
			MysqlInfo: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "region.abc.com:123",
					Username: "user",
					Password: "raw-password",
				},
				PrivatelinkInfo: &dataflow.PrivatelinkInfo{
					DfPrivatelinkEndpointId: 223,
				},
			},
		}
		err := s.serverlessAdaptor.CompleteEditParams(context.Background(), editParams, currentParamsStr)
		s.Error(err)
		s.Contains(err.Error(), "is invalid")
	})

	s.Run("password is empty", func() {
		editParams := &dataflow.ServerlessTidbInfo{
			ClusterId: 111,
			MysqlInfo: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "region.abc.com:123",
					Username: "user",
					Password: "",
				},
				PrivatelinkInfo: &dataflow.PrivatelinkInfo{
					DfPrivatelinkEndpointId: 222,
				},
			},
		}
		err := s.serverlessAdaptor.CompleteEditParams(context.Background(), editParams, currentParamsStr)
		s.NoError(err)
		s.Equal(currentParams.MysqlInfo.Connection.Password, editParams.MysqlInfo.Connection.Password)
	})

	editParams := &dataflow.ServerlessTidbInfo{
		ClusterId: 111,
		MysqlInfo: &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Endpoint: "region.abc.com:123",
				Username: "user",
				Password: "raw-password",
			},
			PrivatelinkInfo: &dataflow.PrivatelinkInfo{
				DfPrivatelinkEndpointId: 222,
			},
		},
	}
	s.Run("encrypt password error", func() {
		encryptErr := bizerr.New(bizerr.ErrCodeUnknown, "encrypt error")
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", mock.Anything).
			Return(encryptErr)
		defer call.Unset()
		err := s.serverlessAdaptor.CompleteEditParams(context.Background(), editParams, currentParamsStr)
		s.Error(err)
		s.Contains(err.Error(), "encrypt password")
	})

	encryptedPassword := "encrypted-edit-password"
	call := s.mockEncryptor.
		On("EncryptBase64Inplace", &editParams.MysqlInfo.Connection.Password).
		Return(nil).
		Run(func(args mock.Arguments) {
			password := args.Get(0).(*string)
			*password = encryptedPassword
		})
	defer call.Unset()
	s.Run("complete edit params successfully", func() {
		err := s.serverlessAdaptor.CompleteEditParams(context.Background(), editParams, currentParamsStr)
		s.NoError(err)
		s.Equal(encryptedPassword, editParams.MysqlInfo.Connection.Password)
	})
}

func TestServerlessDownstreamAdaptorTestSuite(t *testing.T) {
	suite.Run(t, new(serverlessDownstreamAdaptorTestSuite))
}