package impl

import (
	"context"
	"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"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	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 gcsDownstreamAdaptorTestSuite struct {
	suite.Suite
	gcsAdaptor *gcsDownstreamAdaptor

	mockPreCheckCli         *regionalapi_mock.TiCDCPreCheckClient
	mockTiDBClusterInfraCli *grpcli_mock.TiDBClusterInfraClient
}

func (s *gcsDownstreamAdaptorTestSuite) SetupSuite() {
	s.mockPreCheckCli = regionalapi_mock.NewTiCDCPreCheckClient(s.T())
	s.mockTiDBClusterInfraCli = grpcli_mock.NewTiDBClusterInfraClient(s.T())

	s.gcsAdaptor = &gcsDownstreamAdaptor{
		cdcPreCheckCli:      s.mockPreCheckCli,
		tidbClusterInfraCli: s.mockTiDBClusterInfraCli,
	}
}

func (s *gcsDownstreamAdaptorTestSuite) TestParseGCSParams() {
	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.GcsInfo{},
			expectedError:    true,
		},
		{
			name:             "right type param",
			downstreamParams: &dataflow.GcsInfo{},
			expectedError:    false,
		},
	}

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

func (s *gcsDownstreamAdaptorTestSuite) TestParseGCSParamsStr() {
	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.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "gs://bucket/key",
				},
			},
			expectedError: false,
		},
	}

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

func (s *gcsDownstreamAdaptorTestSuite) 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.gcsAdaptor.TestConnection(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

	mockDownstreamParams := &dataflow.GcsInfo{
		Endpoint: &dataflow.GcsEndpoint{
			GcsUri: "gs://bucket/key",
		},
	}

	s.Run("get clsuter role error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockTiDBClusterInfraCli.
			On("GetClusterNodeRole", mock.Anything, mock.Anything).
			Return("", mockErr)
		defer call.Unset()
		err := s.gcsAdaptor.TestConnection(context.Background(), cluster, mockDownstreamParams)
		s.Error(err)
	})

	mockRole := "role-1"
	clusterRoleCall := s.mockTiDBClusterInfraCli.
		On("GetClusterNodeRole", mock.Anything, cluster.Id).
		Return(mockRole, nil)
	defer clusterRoleCall.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.gcsAdaptor.TestConnection(context.Background(), cluster, mockDownstreamParams)
		s.Error(err)
	})

	expectedRegion := &regionalapi.Region{
		Name:          "us-west-2",
		TenantID:      1,
		CloudProvider: "aws",
	}
	expectedReq := &ticdc_v1.TestNetworkConnectionRequest{
		ClusterId: "111",
		Type:      "gcs",
		GcsParam: &ticdc_v1.TestNetworkConnectionRequest_GCSParameter{
			GcsUrl:                    "gs://bucket/key",
			Region:                    "us-west-2",
			ImpersonateServiceAccount: mockRole,
		},
	}
	call := s.mockPreCheckCli.
		On("TestNetworkConnection", mock.Anything, expectedRegion, expectedReq).
		Return(nil)
	defer call.Unset()
	s.Run("test connection successfully", func() {
		err := s.gcsAdaptor.TestConnection(context.Background(), cluster, mockDownstreamParams)
		s.NoError(err)
	})
}

func (s *gcsDownstreamAdaptorTestSuite) 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.gcsAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

	s.Run("parse start tso error", func() {
		downstreamParams := &dataflow.GcsInfo{
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_UTC,
				Utc:  time.Now().Format(time.RFC1123),
			},
		}
		err := s.gcsAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

	downstreamParams := &dataflow.GcsInfo{
		StartPosition: &dataflow.StartPosition{
			Mode: dataflow.StartMode_FROM_TSO,
			Tso:  123123,
		},
	}
	s.Run("pre check error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockPreCheckCli.
			On("ValidateChangefeedParameters", mock.Anything, mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()

		err := s.gcsAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

	s.Run("pre check successfully", func() {
		expectedRegion := &regionalapi.Region{
			Name:          "us-west-2",
			TenantID:      1,
			CloudProvider: "aws",
		}
		expectedReq := &ticdc_v1.ValidateChangefeedParametersRequest{
			ClusterId: "111",
			Type:      "gcs",
			StartTso:  123123,
		}
		call := s.mockPreCheckCli.
			On("ValidateChangefeedParameters", mock.Anything, expectedRegion, expectedReq).
			Return(nil)
		defer call.Unset()
		s.Run("pre check successfully", func() {
			err := s.gcsAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
			s.NoError(err)
		})
	})
}

func (s *gcsDownstreamAdaptorTestSuite) TestAssembleURIConfig() {
	gcsParams := &dataflow.GcsInfo{
		Endpoint: &dataflow.GcsEndpoint{
			GcsUri: "gs://bucket/key",
		},
		StorageFlushConfig: &dataflow.StorageFlushConfig{
			FlushInterval: "5m",
			FileSize:      1000,
		},
		DataFormat: &dataflow.S3Dataformat{
			Protocol:            dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
			EnableTidbExtension: true,
		},
	}

	expectedURIConfig := &domain.TiCDCChangefeedGcsURIConfigData{
		GcsURI:               "gs://bucket/key",
		ForcePathStyle:       pointer.ToBool(true),
		FlushInterval:        "5m",
		FileSize:             1000,
		TransactionAtomicity: pointer.ToString(defaultTransactionAtomicity),
		EnableTiDBExtension:  pointer.ToBool(true),
	}

	uriConfig, err := s.gcsAdaptor.assembleURIConfig(gcsParams)
	s.NoError(err)
	s.Equal(expectedURIConfig, uriConfig)
}

func (s *gcsDownstreamAdaptorTestSuite) TestAssembleChangefeedConfig() {
	cluster := &core.Cluster{}
	bizCf := &entity.BizChangefeed{}
	s.Run("parse gcs params error", func() {
		downstreamParams := &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Endpoint: "abc:3306",
			},
		}
		downstreamParamsBytes, err := json.Marshal(downstreamParams)
		s.NoError(err)
		bizCf.DownstreamParams = string(downstreamParamsBytes)
		_, err = s.gcsAdaptor.AssembleChangefeedConfig(context.Background(), cluster, bizCf)
		s.Error(err)
		s.Contains(err.Error(), "unmarshal gcs params")
	})

	s.Run("assemble changefeed config error", func() {
		downstreamParams := &dataflow.GcsInfo{
			Endpoint: &dataflow.GcsEndpoint{
				GcsUri: "gs://bucket/key",
			},
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_UTC,
				Utc:  time.Now().Format(time.RFC1123),
			},
		}
		downstreamParamsBytes, err := json.Marshal(downstreamParams)
		s.NoError(err)
		bizCf.DownstreamParams = string(downstreamParamsBytes)
		_, err = s.gcsAdaptor.AssembleChangefeedConfig(context.Background(), cluster, bizCf)
		s.Error(err)
		s.Contains(err.Error(), "invalid start tso")
	})

	s.Run("assemble changefeed config successfully", func() {
		downstreamParams := &dataflow.GcsInfo{
			Endpoint: &dataflow.GcsEndpoint{
				GcsUri: "gs://bucket/key",
			},
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_TSO,
				Tso:  123123123,
			},
			DataFormat: &dataflow.S3Dataformat{
				Terminator:        ";",
				DateSeparator:     "day",
				Protocol:          dataflow.S3Protocol_S3_PROTOCOL_CSV,
				OutputFieldHeader: true,
			},
			StorageFlushConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "5m",
				FileSize:      1000,
			},
			TableConfig: &dataflow.TableConfig{
				FilterRule: []string{"*.*"},
				EventFilterRule: []*dataflow.EventFilterRule{
					{
						Matcher:     []string{"test.*"},
						IgnoreEvent: []string{"delete"},
					},
				},
				CaseSensitive: true,
				Mode:          dataflow.TableMode_FORCE_SYNC,
			},
		}
		expectedConfig := &domain.TiCDCChangefeedConfig{
			URIConfig: &domain.TiCDCChangefeedGcsURIConfigData{
				GcsURI:               "gs://bucket/key",
				ForcePathStyle:       pointer.ToBool(true),
				FlushInterval:        "5m",
				FileSize:             1000,
				TransactionAtomicity: pointer.ToString(defaultTransactionAtomicity),
			},
			Config: &domain.TiCDCChangefeedConfigData{
				StartTs: pointer.ToUint64(123123123),
				ReplicaConfig: &clusterproxy.ReplicaConfig{
					CaseSensitive:         pointer.ToBool(true),
					ForceReplicate:        pointer.ToBool(true),
					IgnoreIneligibleTable: pointer.ToBool(false),
					Sink: &clusterproxy.SinkConfigV2{
						Terminator:    ";",
						DateSeparator: "day",
						Protocol:      "csv",
						CSVConfig: &clusterproxy.CSVConfig{
							Delimiter:            ",",
							Quote:                "\"",
							NullString:           "\\N",
							IncludeCommitTs:      pointer.ToBool(false),
							BinaryEncodingMethod: BinaryEncodingMethodBase64,
							OutputFieldHeader:    pointer.ToBool(true),
						},
						CloudStorageConfig: &clusterproxy.CloudStorageConfig{
							OutputRawChangeEvent: pointer.ToBool(false),
						},
					},
					Filter: &clusterproxy.FilterConfig{
						Rules: []string{"*.*"},
						EventFilters: []clusterproxy.EventFilterRule{
							{
								Matcher:     []string{"test.*"},
								IgnoreEvent: []string{"delete"},
							},
						},
					},
				},
			},
		}
		downstreamParamsBytes, err := json.Marshal(downstreamParams)
		s.NoError(err)
		bizCf.DownstreamParams = string(downstreamParamsBytes)
		changefeedConfig, err := s.gcsAdaptor.AssembleChangefeedConfig(context.Background(), cluster, bizCf)
		s.NoError(err)
		s.Equal(expectedConfig, changefeedConfig)
	})
}

func (s *gcsDownstreamAdaptorTestSuite) TestCompleteEditParams() {
	invalidCurrentParamsBytes, _ := json.Marshal(&dataflow.MySQLInfo{
		Connection: &dataflow.Connection{
			Endpoint: "abc:3306",
		},
	})
	currentParamsBytes, _ := json.Marshal(&dataflow.GcsInfo{
		Endpoint: &dataflow.GcsEndpoint{
			GcsUri: "gs://bucket/key",
		},
	})
	s.Run("invalid edit params", func() {
		editParams := &dataflow.MySQLInfo{}
		err := s.gcsAdaptor.CompleteEditParams(context.Background(), editParams, string(currentParamsBytes))
		s.Error(err)
		s.Contains(err.Error(), "downstreamParams is not *dataflow.GcsInfo type")
	})

	s.Run("invalid current params", func() {
		editParams := &dataflow.GcsInfo{}
		err := s.gcsAdaptor.CompleteEditParams(context.Background(), editParams, string(invalidCurrentParamsBytes))
		s.Error(err)
		s.Contains(err.Error(), "unmarshal gcs params error")
	})

	s.Run("endpoint is changed", func() {
		editParams := &dataflow.GcsInfo{
			Endpoint: &dataflow.GcsEndpoint{
				GcsUri: "gs://bucket-2/key",
			},
		}
		err := s.gcsAdaptor.CompleteEditParams(context.Background(), editParams, string(currentParamsBytes))
		s.NoError(err)
		s.Equal("gs://bucket/key", editParams.Endpoint.GcsUri)
	})
}

func TestGcsDownstreamAdaptorTestSuite(t *testing.T) {
	suite.Run(t, new(gcsDownstreamAdaptorTestSuite))
}