package gcp

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 ExecuteTidbToGCS(t *testing.T) {
	uniqueKey := fmt.Sprintf("%s_%s",
		dpCommon.SCHEMA_TIDB_TO_GCS, util.FnvHash(common.Envn+string(dpCommon.CloudProviderTypeGCP)+common.GetDbassRequestInfo(t).CommonEnv.Email))
	test := &CdcTidbToGCSSceneTest{
		cdc.CommonTidbToGCSSceneTest{
			CommonSceneTest: cdc.CommonSceneTest{
				SceneTest: dpCommon.SceneTest{
					T:                t,
					DbaasRequestInfo: common.GetDbassRequestInfo(t),
					ClusterId:        GetClusterId(),
					TestSchemaName:   uniqueKey,
				},
				StartUTC: time.Now().UTC(),
			},
		},
	}
	test.Run()
}

type CdcTidbToGCSSceneTest struct {
	cdc.CommonTidbToGCSSceneTest
}

func (s *CdcTidbToGCSSceneTest) Run() {
	allure.Test(s.T,
		allure.Epic(fmt.Sprintf("%s-%s", dpCommon.CloudProviderTypeGCP, dpCommon.TidbToGCS)),
		allure.Feature(string(dpCommon.TidbToGCS)),
		allure.Description(string(dpCommon.TidbToGCS)),
		allure.Action(func() {
			s.GetAccountId()
			s.ExecuteManagerCommandToGCSStep(s.InitGCS)
			s.AddBucketRole(s.T)
			s.TestDownStreamConnectionStep(s.buildTestConnectToGCSReq())
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.InitTidb)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.SetGC720h)
			s.DescribeTidbSchemaTableStep()
			s.CreateChangefeedStep(s.buildCreateTidbToGCSChangefeedReq())
			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.buildEditTidbToGCSChangefeedReq())
			s.WaitChangefeedStatusReadyStep(models.DataflowConnectorStatusPAUSED)
			s.StartChangefeedStep()
			s.WaitChangefeedStatusReadyStep(models.DataflowConnectorStatusRUNNING)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.CreateIncreStageTable)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.InsertToIncreStageTable)
			s.CheckGCSBucketSize(s.T)
			s.DeleteChangefeedStep()
		}))
}

func (s *CdcTidbToGCSSceneTest) buildTestConnectToGCSReq() *models.DataflowConnectorCreateReq {
	return &models.DataflowConnectorCreateReq{
		ConnectorType: models.DataflowConnectorTypeGCS.Pointer(),
		GCSInfo: &models.DataflowGCSInfo{
			EndPoint: &models.DataflowEndPoint{
				GcsUri: s.DbaasRequestInfo.CommonEnv.GcsUri,
			},
		},
	}
}

func (s *CdcTidbToGCSSceneTest) buildCreateTidbToGCSChangefeedReq() *models.DataflowConnectorCreateReq {
	return &models.DataflowConnectorCreateReq{
		ConnectorType: models.DataflowConnectorTypeGCS.Pointer(),
		GCSInfo: &models.DataflowGCSInfo{
			EndPoint: &models.DataflowEndPoint{
				GcsUri: s.DbaasRequestInfo.CommonEnv.GcsUri,
			},
			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.GcsUri),
		ChangefeedInstanceType: dpCommon.RCU2,
	}
}

func (s *CdcTidbToGCSSceneTest) buildEditTidbToGCSChangefeedReq() *models.DataflowConnectorEditReq {
	return &models.DataflowConnectorEditReq{
		ConnectorID:   s.ChangefeedId,
		ConnectorType: models.DataflowConnectorTypeGCS.Pointer(),
		GCSInfo: &models.DataflowGCSInfo{
			EndPoint: &models.DataflowEndPoint{
				GcsUri: s.DbaasRequestInfo.CommonEnv.GcsUri,
			},
			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",
			},
		},
	}
}