package aws

import (
	"fmt"
	"testing"
	"time"

	"github.com/dailymotion/allure-go"
	"github.com/tidbcloud/qa-PlatformAPIAuto/common"
	"github.com/tidbcloud/qa-PlatformAPIAuto/models"
	"github.com/tidbcloud/qa-PlatformAPIAuto/util"
	"github.com/tidbcloud/qa-PlatformAPIAuto/webScene/cdc"
	dpCommon "github.com/tidbcloud/qa-PlatformAPIAuto/webScene/common"
)

func ExecuteTidbToKafka(t *testing.T) {
	uniqueKey := fmt.Sprintf("%s_%s",
		dpCommon.SCHEMA_TIDB_TO_KAFKA, util.FnvHash(common.Envn+string(dpCommon.CloudProviderTypeAWS)+common.GetDbassRequestInfo(t).CommonEnv.Email))
	test := &CdcTidbToKafkaSceneTest{
		cdc.CommonTidbToKafkaSceneTest{
			CommonSceneTest: cdc.CommonSceneTest{
				SceneTest: dpCommon.SceneTest{
					T:                t,
					DbaasRequestInfo: common.GetDbassRequestInfo(t),
					ClusterId:        GetClusterId(),
					TestSchemaName:   uniqueKey,
				},
				StartUTC: time.Now().UTC(),
			},
			Topic:                uniqueKey,
			GroupID:              uniqueKey,
			FullStageCheckCount:  0,
			IncreStageCheckCount: 0,
		},
	}
	test.Run()
}

type CdcTidbToKafkaSceneTest struct {
	cdc.CommonTidbToKafkaSceneTest
}

func (s *CdcTidbToKafkaSceneTest) Run() {
	allure.Test(s.T,
		allure.Epic(fmt.Sprintf("%s-%s", dpCommon.CloudProviderTypeAWS, dpCommon.TidbToKafka)),
		allure.Feature(string(dpCommon.TidbToKafka)),
		allure.Description(string(dpCommon.TidbToKafka)),
		allure.Action(func() {
			s.TestDownStreamConnectionStep(s.buildTestConnectToKafkaReq())
			s.ExecuteManagerCommandToKafkaStep(s.InitKafka)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.InitTidb)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.SetGC720h)
			s.DescribeTidbSchemaTableStep()
			s.CreateChangefeedStep(s.buildCreateTidbToKafkaChangefeedReq())
			s.WaitChangefeedStatusReadyStep(models.DataflowConnectorStatusRUNNING)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.SetGC10m)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.InsertToFullStageTable)
			// comment out read kafka step since the case driver network may not able to access kafka
			// s.ReadKafkaDataWithRetryStep(s.CheckFullStageKafkaData)
			s.StopChangefeedStep()
			s.WaitChangefeedStatusReadyStep(models.DataflowConnectorStatusPAUSED)
			s.EditChangefeedStep(s.buildEditTidbToKafkaChangefeedReq())
			s.WaitChangefeedStatusReadyStep(models.DataflowConnectorStatusPAUSED)
			s.StartChangefeedStep()
			s.WaitChangefeedStatusReadyStep(models.DataflowConnectorStatusRUNNING)
			//s.ExecuteTidbDataWithDefaultClusterIDStep(s.CreateIncreStageTable)
			//s.ExecuteTidbDataWithDefaultClusterIDStep(s.InsertToIncreStageTable)
			//s.ReadKafkaDataWithRetryStep(s.CheckIncreStageKafkaData)
			s.DeleteChangefeedStep()
		}))
}

func (s *CdcTidbToKafkaSceneTest) buildTestConnectToKafkaReq() *models.DataflowConnectorCreateReq {
	return &models.DataflowConnectorCreateReq{
		ConnectorType: models.DataflowConnectorTypeKAFKA.Pointer(),
		KafkaInfo: &models.DataflowKafkaInfo{
			Broker: &models.DataflowBroker{
				BrokerEndpoints: s.DbaasRequestInfo.CommonEnv.CdcKafkaUrl,
				KafkaVersion:    models.DataflowKafkaVersionKAFKAVERSION2XX.Pointer(),
				Compression:     models.DataflowCompressionTypeNONE.Pointer(),
				TLSEnable:       false,
			},
		},
	}
}

func (s *CdcTidbToKafkaSceneTest) buildCreateTidbToKafkaChangefeedReq() *models.DataflowConnectorCreateReq {
	return &models.DataflowConnectorCreateReq{
		ConnectorType: models.DataflowConnectorTypeKAFKA.Pointer(),
		KafkaInfo: &models.DataflowKafkaInfo{
			Broker: &models.DataflowBroker{
				BrokerEndpoints: s.DbaasRequestInfo.CommonEnv.CdcKafkaUrl,
				KafkaVersion:    models.DataflowKafkaVersionKAFKAVERSION2XX.Pointer(),
				Compression:     models.DataflowCompressionTypeNONE.Pointer(),
				TLSEnable:       false,
			},
			Authentication: &models.DataflowAuthentication{
				AuthType: models.DataflowKafkaAuthTypeDISABLE.Pointer(),
			},
			DataFormat: &models.DataflowDataFormat{
				Protocol:            models.DataflowProtocolCANALJSON.Pointer(),
				EnableTidbExtension: false,
			},
			TopicPartitionConfig: &models.DataflowTopicPartitionConfig{
				DispatchType:        models.DataflowDispatchTypeDISPATCHTYPEONETOPIC.Pointer(),
				DefaultTopic:        s.Topic,
				PartitionDispatcher: models.DataflowPartitionDispatcherINDEXVALUE.Pointer(),
				ReplicationFactor:   1,
				PartitionNum:        1,
			},
			TableConfig: &models.DataflowTableConfig{
				FilterRule: []string{
					fmt.Sprintf("%s.test*", s.TestSchemaName),
				},
				Mode: models.DataflowTableModeFORCESYNC.Pointer(),
			},
			StartPosition: &models.DataflowStartPosition{
				Mode: models.DataflowStartModeFROMNOW.Pointer(),
			},
		},
		ChangefeedInstanceType: dpCommon.RCU2,
	}
}

func (s *CdcTidbToKafkaSceneTest) buildEditTidbToKafkaChangefeedReq() *models.DataflowConnectorEditReq {
	return &models.DataflowConnectorEditReq{
		ConnectorID:   s.ChangefeedId,
		ConnectorType: models.DataflowConnectorTypeKAFKA.Pointer(),
		KafkaInfo: &models.DataflowKafkaInfo{
			Broker: &models.DataflowBroker{
				BrokerEndpoints: s.DbaasRequestInfo.CommonEnv.CdcKafkaUrl,
				KafkaVersion:    models.DataflowKafkaVersionKAFKAVERSION2XX.Pointer(),
				Compression:     models.DataflowCompressionTypeNONE.Pointer(),
				TLSEnable:       false,
			},
			Authentication: &models.DataflowAuthentication{
				AuthType: models.DataflowKafkaAuthTypeDISABLE.Pointer(),
			},
			DataFormat: &models.DataflowDataFormat{
				Protocol:            models.DataflowProtocolCANALJSON.Pointer(),
				EnableTidbExtension: false,
			},
			TopicPartitionConfig: &models.DataflowTopicPartitionConfig{
				DispatchType:        models.DataflowDispatchTypeDISPATCHTYPEONETOPIC.Pointer(),
				DefaultTopic:        s.Topic,
				PartitionDispatcher: models.DataflowPartitionDispatcherINDEXVALUE.Pointer(),
				ReplicationFactor:   1,
				PartitionNum:        1,
			},
			TableConfig: &models.DataflowTableConfig{
				FilterRule: []string{
					fmt.Sprintf("%s.test*", s.TestSchemaName),
					fmt.Sprintf("%s.incre_test*", s.TestSchemaName),
				},
				Mode: models.DataflowTableModeFORCESYNC.Pointer(),
			},
		},
	}
}