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"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	encryptor_mock "github.com/tidbcloud/dataflow-service/pkg/encryptor-wrapper/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 azureBlobStorageDownstreamAdaptorTestSuite struct {
	suite.Suite
	azBlobAdaptor *azureBlobStorageDownstreamAdaptor

	mockPreCheckCli *regionalapi_mock.TiCDCPreCheckClient
	mockEncryptor   *encryptor_mock.Encryptor
}

func (s *azureBlobStorageDownstreamAdaptorTestSuite) SetupSuite() {
	s.mockPreCheckCli = regionalapi_mock.NewTiCDCPreCheckClient(s.T())
	s.mockEncryptor = encryptor_mock.NewEncryptor(s.T())

	s.azBlobAdaptor = &azureBlobStorageDownstreamAdaptor{
		cdcPreCheckCli: s.mockPreCheckCli,
		encryptor:      s.mockEncryptor,
	}
}

func (s *azureBlobStorageDownstreamAdaptorTestSuite) TestParseAzureBlobStorageParams() {
	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.AzureBlobStorageInfo{},
			expectedError:    true,
		},
		{
			name:             "right type param",
			downstreamParams: &dataflow.AzureBlobStorageInfo{},
			expectedError:    false,
		},
		{
			name: "invalid date separator",
			downstreamParams: &dataflow.AzureBlobStorageInfo{
				DataFormat: &dataflow.S3Dataformat{DateSeparator: "week"},
			},
			expectedError: true,
		},
	}

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

func (s *azureBlobStorageDownstreamAdaptorTestSuite) TestParseAzureBlobStorageParamsStr() {
	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.AzureBlobStorageInfo{
				Endpoint: &dataflow.AzureBlobStorageEndpoint{
					BlobUri:  "https://storageaccount123.blob.core.windows.net/container123",
					SasToken: "token123",
				},
			},
			expectedError: false,
		},
		{
			name: "invalid date separator",
			downstreamParams: &dataflow.AzureBlobStorageInfo{
				DataFormat: &dataflow.S3Dataformat{DateSeparator: "week"},
			},
			expectedError: true,
		},
	}

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

func (s *azureBlobStorageDownstreamAdaptorTestSuite) TestTestConnection() {
	cluster := &core.Cluster{
		Id:                111,
		ProjectId:         11,
		TenantId:          1,
		RegionName:        "eastus2",
		InfraProviderType: "aks",
	}

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

	mockDownstreamParams := &dataflow.AzureBlobStorageInfo{
		Endpoint: &dataflow.AzureBlobStorageEndpoint{
			BlobUri:  "https://storageaccount123.blob.core.windows.net/container123",
			SasToken: "token123",
		},
	}

	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.azBlobAdaptor.TestConnection(context.Background(), cluster, mockDownstreamParams)
		s.Error(err)
	})

	expectedRegion := &regionalapi.Region{
		Name:          "eastus2",
		TenantID:      1,
		CloudProvider: "azure",
	}
	expectedReq := &ticdc_v1.TestNetworkConnectionRequest{
		ClusterId: "111",
		Type:      "azblob",
		AzureBlobStorageParam: &ticdc_v1.TestNetworkConnectionRequest_AzureBlobStorageParameter{
			BlobUri:  "https://storageaccount123.blob.core.windows.net/container123",
			SasToken: "token123",
		},
	}
	call := s.mockPreCheckCli.
		On("TestNetworkConnection", mock.Anything, expectedRegion, expectedReq).
		Return(nil)
	defer call.Unset()
	s.Run("test connection successfully", func() {
		err := s.azBlobAdaptor.TestConnection(context.Background(), cluster, mockDownstreamParams)
		s.NoError(err)
	})
}

func (s *azureBlobStorageDownstreamAdaptorTestSuite) TestPreCheckChangefeedParameters() {
	cluster := &core.Cluster{
		Id:                111,
		ProjectId:         11,
		TenantId:          1,
		RegionName:        "eastus2",
		InfraProviderType: "aks",
	}

	s.Run("downstream param error", func() {
		downstreamParams := &dataflow.MySQLInfo{}
		err := s.azBlobAdaptor.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.azBlobAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

	downstreamParams := &dataflow.AzureBlobStorageInfo{
		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.azBlobAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

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

func (s *azureBlobStorageDownstreamAdaptorTestSuite) TestEncryptPasswordInPlace() {
	s.Run("downstream param error", func() {
		downstreamParams := &dataflow.MySQLInfo{}
		err := s.azBlobAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.Error(err)
	})

	s.Run("sas token nil", func() {
		downstreamParams := &dataflow.AzureBlobStorageInfo{}
		err := s.azBlobAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.NoError(err)
	})

	downstreamParams := &dataflow.AzureBlobStorageInfo{
		Endpoint: &dataflow.AzureBlobStorageEndpoint{
			BlobUri:  "https://storageaccount123.blob.core.windows.net/container123",
			SasToken: "",
		},
	}
	s.Run("sas token empty", func() {
		err := s.azBlobAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.NoError(err)
	})

	downstreamParams.Endpoint.SasToken = "token123"
	s.Run("encrypt sas token error", func() {
		encryptErr := bizerr.New(bizerr.ErrCodeUnknown, "encrypt sas token error")
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", mock.Anything).
			Return(encryptErr)
		defer call.Unset()

		err := s.azBlobAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.Error(err)
		s.Contains(err.Error(), encryptErr.ErrMsg())
	})

	s.Run("encrypt successfully", func() {
		encryptedSASToken := base64.StdEncoding.EncodeToString([]byte(downstreamParams.Endpoint.SasToken))
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", &downstreamParams.Endpoint.SasToken).
			Return(nil).
			Run(func(args mock.Arguments) {
				token := args.Get(0).(*string)
				*token = encryptedSASToken
			})
		defer call.Unset()

		err := s.azBlobAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.NoError(err)
		s.Equal(encryptedSASToken, downstreamParams.Endpoint.SasToken)
	})
}

func (s *azureBlobStorageDownstreamAdaptorTestSuite) TestAssembleURIConfig() {
	azBlobParams := &dataflow.AzureBlobStorageInfo{
		Endpoint: &dataflow.AzureBlobStorageEndpoint{
			BlobUri:  "https://storageaccount123.blob.core.windows.net/container123",
			SasToken: "encrypted-token",
		},
		StorageFlushConfig: &dataflow.StorageFlushConfig{
			FlushInterval: "5m",
			FileSize:      1000,
		},
		DataFormat: &dataflow.S3Dataformat{
			Protocol:            dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
			EnableTidbExtension: true,
		},
	}

	s.Run("decrypt sas token error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "decrypt error")
		call := s.mockEncryptor.On("DecryptBase64", mock.Anything).Return("", mockErr)
		defer call.Unset()
		_, err := s.azBlobAdaptor.assembleURIConfig(azBlobParams)
		s.Error(err)
		s.Contains(err.Error(), "decrypt sas token")
	})

	s.Run("assemble uri config successfully", func() {
		expectedSASToken := "token123"
		call := s.mockEncryptor.
			On("DecryptBase64", azBlobParams.Endpoint.SasToken).
			Return(expectedSASToken, nil)
		defer call.Unset()

		expectedURIConfig := &domain.TiCDCChangefeedAzureBlobStorageURIConfigData{
			BlobPath:             "container123",
			AccountName:          "storageaccount123",
			SASToken:             pointer.ToString(expectedSASToken),
			ForcePathStyle:       pointer.ToBool(true),
			FlushInterval:        "5m",
			FileSize:             1000,
			TransactionAtomicity: pointer.ToString(defaultTransactionAtomicity),
			EnableTiDBExtension:  pointer.ToBool(true),
		}

		uriConfig, err := s.azBlobAdaptor.assembleURIConfig(azBlobParams)
		s.NoError(err)
		s.Equal(expectedURIConfig, uriConfig)
	})
}

func (s *azureBlobStorageDownstreamAdaptorTestSuite) TestAssembleChangefeedConfig() {
	cluster := &core.Cluster{}
	bizCf := &entity.BizChangefeed{}
	s.Run("parse azure blob storage 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.azBlobAdaptor.AssembleChangefeedConfig(context.Background(), cluster, bizCf)
		s.Error(err)
		s.Contains(err.Error(), "unmarshal azure blob storage params")
	})

	s.Run("decrypt sas token error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "decrypt error")
		call := s.mockEncryptor.On("DecryptBase64", mock.Anything).Return("", mockErr)
		defer call.Unset()

		downstreamParams := &dataflow.AzureBlobStorageInfo{
			Endpoint: &dataflow.AzureBlobStorageEndpoint{
				BlobUri:  "https://storageaccount123.blob.core.windows.net/container123",
				SasToken: "encrypted-token",
			},
		}
		downstreamParamsBytes, err := json.Marshal(downstreamParams)
		s.NoError(err)
		bizCf.DownstreamParams = string(downstreamParamsBytes)
		_, err = s.azBlobAdaptor.AssembleChangefeedConfig(context.Background(), cluster, bizCf)
		s.Error(err)
		s.Contains(err.Error(), "decrypt sas token")
	})

	expectedSASToken := "token123"
	decryptCall := s.mockEncryptor.On("DecryptBase64", mock.Anything).Return(expectedSASToken, nil)
	defer decryptCall.Unset()

	s.Run("assemble changefeed config error", func() {
		downstreamParams := &dataflow.AzureBlobStorageInfo{
			Endpoint: &dataflow.AzureBlobStorageEndpoint{
				BlobUri:  "https://storageaccount123.blob.core.windows.net/container123",
				SasToken: "encrypted-token",
			},
			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.azBlobAdaptor.AssembleChangefeedConfig(context.Background(), cluster, bizCf)
		s.Error(err)
		s.Contains(err.Error(), "invalid start tso")
	})

	s.Run("assemble changefeed config successfully", func() {
		downstreamParams := &dataflow.AzureBlobStorageInfo{
			Endpoint: &dataflow.AzureBlobStorageEndpoint{
				BlobUri:  "https://storageaccount123.blob.core.windows.net/container123",
				SasToken: "encrypted-token",
			},
			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.TiCDCChangefeedAzureBlobStorageURIConfigData{
				BlobPath:             "container123",
				AccountName:          "storageaccount123",
				SASToken:             pointer.ToString(expectedSASToken),
				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.azBlobAdaptor.AssembleChangefeedConfig(context.Background(), cluster, bizCf)
		s.NoError(err)
		s.Equal(expectedConfig, changefeedConfig)
	})
}

func (s *azureBlobStorageDownstreamAdaptorTestSuite) TestCompleteEditParams() {
	invalidCurrentParamsBytes, _ := json.Marshal(&dataflow.MySQLInfo{
		Connection: &dataflow.Connection{
			Endpoint: "abc:3306",
		},
	})
	currentParamsBytes, _ := json.Marshal(&dataflow.AzureBlobStorageInfo{
		Endpoint: &dataflow.AzureBlobStorageEndpoint{
			BlobUri:  "https://storageaccount123.blob.core.windows.net/container123",
			SasToken: "encrypted-token-1",
		},
	})
	s.Run("invalid edit params", func() {
		editParams := &dataflow.MySQLInfo{}
		err := s.azBlobAdaptor.CompleteEditParams(context.Background(), editParams, string(currentParamsBytes))
		s.Error(err)
		s.Contains(err.Error(), "downstreamParams is not *dataflow.AzureBlobStorageInfo type")
	})

	s.Run("invalid current params", func() {
		editParams := &dataflow.AzureBlobStorageInfo{}
		err := s.azBlobAdaptor.CompleteEditParams(context.Background(), editParams, string(invalidCurrentParamsBytes))
		s.Error(err)
		s.Contains(err.Error(), "unmarshal azure blob storage params error")
	})

	s.Run("endpoint is changed", func() {
		editParams := &dataflow.AzureBlobStorageInfo{
			Endpoint: &dataflow.AzureBlobStorageEndpoint{
				BlobUri:  "https://storageaccount123.blob.core.windows.net/container456",
				SasToken: "",
			},
		}
		err := s.azBlobAdaptor.CompleteEditParams(context.Background(), editParams, string(currentParamsBytes))
		s.NoError(err)
		s.Equal("https://storageaccount123.blob.core.windows.net/container123", editParams.Endpoint.BlobUri)
		s.Equal("encrypted-token-1", editParams.Endpoint.SasToken)
	})

	s.Run("sas token is changed but encrypt error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "encrypt error")
		call := s.mockEncryptor.On("EncryptBase64Inplace", mock.Anything).Return(mockErr)
		defer call.Unset()

		editParams := &dataflow.AzureBlobStorageInfo{
			Endpoint: &dataflow.AzureBlobStorageEndpoint{
				BlobUri:  "https://storageaccount123.blob.core.windows.net/container123",
				SasToken: "token456",
			},
		}
		err := s.azBlobAdaptor.CompleteEditParams(context.Background(), editParams, string(currentParamsBytes))
		s.Error(err)
		s.Contains(err.Error(), "encrypt sas token")
	})

	s.Run("sas token is changed and complete successfully", func() {
		expectedToken := "encrypted-token-2"
		editParams := &dataflow.AzureBlobStorageInfo{
			Endpoint: &dataflow.AzureBlobStorageEndpoint{
				BlobUri:  "https://storageaccount123.blob.core.windows.net/container123",
				SasToken: "token456",
			},
		}
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", &editParams.Endpoint.SasToken).
			Return(nil).
			Run(func(args mock.Arguments) {
				token := args.Get(0).(*string)
				*token = expectedToken
			})
		defer call.Unset()
		err := s.azBlobAdaptor.CompleteEditParams(context.Background(), editParams, string(currentParamsBytes))
		s.NoError(err)
		s.Equal(expectedToken, editParams.Endpoint.SasToken)
	})
}

func TestAzureBlobStorageDownstreamAdaptorTestSuite(t *testing.T) {
	suite.Run(t, new(azureBlobStorageDownstreamAdaptorTestSuite))
}