package aws

import (
	"fmt"
	l4g "github.com/alecthomas/log4go"
	"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 ExecuteTidbToS3(t *testing.T) {
	uniqueKey := fmt.Sprintf("%s_%s",
		dpCommon.SCHEMA_TIDB_TO_S3, util.FnvHash(common.Envn+string(dpCommon.CloudProviderTypeAWS)+common.GetDbassRequestInfo(t).CommonEnv.Email))
	test := &CdcTidbToS3SceneTest{
		cdc.CommonTidbToS3SceneTest{
			CommonSceneTest: cdc.CommonSceneTest{
				SceneTest: dpCommon.SceneTest{
					T:                t,
					DbaasRequestInfo: common.GetDbassRequestInfo(t),
					ClusterId:        GetClusterId(),
					TestSchemaName:   uniqueKey,
				},
				StartUTC: time.Now().UTC(),
			},
		},
	}
	test.Run()
}

type CdcTidbToS3SceneTest struct {
	cdc.CommonTidbToS3SceneTest
}

func (s *CdcTidbToS3SceneTest) Run() {
	allure.Test(s.T,
		allure.Epic(fmt.Sprintf("%s-%s", dpCommon.CloudProviderTypeAWS, dpCommon.TidbToS3)),
		allure.Feature(string(dpCommon.TidbToS3)),
		allure.Description(string(dpCommon.TidbToS3)),
		allure.Action(func() {
			s.TestDownStreamConnectionStep(s.buildTestConnectToS3Req())
			s.ExecuteManagerCommandToS3Step(s.InitS3)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.InitTidb)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.SetGC720h)
			s.DescribeTidbSchemaTableStep()
			s.CreateChangefeedStep(s.buildCreateTidbToS3ChangefeedReq())
			s.WaitChangefeedStatusReadyStep(models.DataflowConnectorStatusRUNNING)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.SetGC10m)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.InitTidb)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.InsertToFullStageTable)
			s.StopChangefeedStep()
			s.WaitChangefeedStatusReadyStep(models.DataflowConnectorStatusPAUSED)
			s.EditChangefeedStep(s.buildEditTidbToS3ChangefeedReq())
			s.WaitChangefeedStatusReadyStep(models.DataflowConnectorStatusPAUSED)
			s.StartChangefeedStep()
			s.WaitChangefeedStatusReadyStep(models.DataflowConnectorStatusRUNNING)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.CreateIncreStageTable)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.InsertToIncreStageTable)
			s.CheckS3BucketSize(s.T)
			s.DeleteChangefeedStep()
		}))
}

func (s *CdcTidbToS3SceneTest) buildTestConnectToS3Req() *models.DataflowConnectorCreateReq {
	l4g.Info("S3Uri:%s", s.DbaasRequestInfo.CommonEnv.S3Uri)
	return &models.DataflowConnectorCreateReq{
		ConnectorType: models.DataflowConnectorTypeS3.Pointer(),
		S3Info: &models.DataflowS3Info{
			S3EndPoint: &models.DataflowS3EndPoint{
				S3Uri:     s.DbaasRequestInfo.CommonEnv.S3Uri,
				AccessKey: s.DbaasRequestInfo.CommonEnv.AccessKey,
				SecretKey: s.DbaasRequestInfo.CommonEnv.SecretKey,
			},
		},
	}
}

func (s *CdcTidbToS3SceneTest) buildCreateTidbToS3ChangefeedReq() *models.DataflowConnectorCreateReq {
	return &models.DataflowConnectorCreateReq{
		ConnectorType: models.DataflowConnectorTypeS3.Pointer(),
		S3Info: &models.DataflowS3Info{
			S3EndPoint: &models.DataflowS3EndPoint{
				S3Uri:     s.DbaasRequestInfo.CommonEnv.S3Uri,
				AccessKey: s.DbaasRequestInfo.CommonEnv.AccessKey,
				SecretKey: s.DbaasRequestInfo.CommonEnv.SecretKey,
			},
			TableConfig: &models.DataflowTableConfig{
				FilterRule: []string{
					fmt.Sprintf("%s.test*", s.TestSchemaName),
					fmt.Sprintf("%s.incre_test*", s.TestSchemaName),
				},
				Mode: models.DataflowTableModeFORCESYNC.Pointer(),
			},
			StartPosition: &models.DataflowStartPosition{
				Mode: models.DataflowStartModeFROMUTC.Pointer(),
				Utc:  s.StartUTC.Format(time.RFC3339),
			},
			DataFormat: &models.DataflowStorageDataFormat{
				BinaryEncodingMethod: "BASE64",
				DateSeparator:        "none",
				Delimiter:            ",",
				IncludeCommitTs:      false,
				Null:                 "\\N",
				Protocol:             "S3_PROTOCOL_CSV",
				Quote:                "\"",
			},
			StorageFlushConfig: &models.DataflowStorageFlushConfig{
				FileSize:      "67108864",
				FlushInterval: "2s",
			},
		},
		//URL: &(s.DbaasRequestInfo.CommonEnv.S3Uri),
		ChangefeedInstanceType: dpCommon.RCU2,
	}
}

func (s *CdcTidbToS3SceneTest) buildEditTidbToS3ChangefeedReq() *models.DataflowConnectorEditReq {
	return &models.DataflowConnectorEditReq{
		ConnectorID:   s.ChangefeedId,
		ConnectorType: models.DataflowConnectorTypeS3.Pointer(),
		S3Info: &models.DataflowS3Info{
			S3EndPoint: &models.DataflowS3EndPoint{
				S3Uri:     s.DbaasRequestInfo.CommonEnv.S3Uri,
				AccessKey: s.DbaasRequestInfo.CommonEnv.AccessKey,
				SecretKey: s.DbaasRequestInfo.CommonEnv.SecretKey,
			},
			TableConfig: &models.DataflowTableConfig{
				FilterRule: []string{
					fmt.Sprintf("%s.test*", s.TestSchemaName),
					fmt.Sprintf("%s.incre_test*", s.TestSchemaName),
				},
				Mode: models.DataflowTableModeFORCESYNC.Pointer(),
			},
			StartPosition: &models.DataflowStartPosition{
				Mode: models.DataflowStartModeFROMNOW.Pointer(),
			},
			DataFormat: &models.DataflowStorageDataFormat{
				BinaryEncodingMethod: "BASE64",
				DateSeparator:        "none",
				Delimiter:            ",",
				IncludeCommitTs:      true,
				Null:                 "\\N",
				Protocol:             "S3_PROTOCOL_CSV",
				Quote:                "\"",
			},
			StorageFlushConfig: &models.DataflowStorageFlushConfig{
				FileSize:      "67108864",
				FlushInterval: "2s",
			},
		},
	}
}