package aws

import (
	"fmt"
	"strconv"
	"testing"
	"time"

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

type CdcTidbToServerlessTiDBSceneTest struct {
	cdc.CommonSceneTest

	ServerlessTidbUser string
	ServerlessTidbPort int32
}

func (s *CdcTidbToServerlessTiDBSceneTest) Run() {
	allure.Test(s.T,
		allure.Epic(fmt.Sprintf("%s-%s", dpCommon.CloudProviderTypeAWS, dpCommon.TidbToServerlessTidb)),
		allure.Feature(string(dpCommon.TidbToServerlessTidb)),
		allure.Description(string(dpCommon.TidbToServerlessTidb)),
		allure.Action(func() {
			s.CreatePrivatelinkEndpoint(s.buildCreatePlEndpointReq())
			s.WaitPrivatelinkEndpointReadyStep()
			s.TestDownStreamConnectionStep(s.buildTestConnectToServerlessTidbReq())
			s.ExecuteServerlessTidbDataStep(s.InitMySQL)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.InitTidb)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.SetGC720h)
			s.DescribeTidbSchemaTableStep()
			s.CreateChangefeedStep(s.buildCreateTidbToServerlessTidbChangefeedReq())
			s.WaitChangefeedStatusReadyStep(models.DataflowConnectorStatusRUNNING)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.SetGC10m)
			// init tidb schema again in order to sync ddl to downstream, otherwise downstream will not have the table and the test will fail
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.InitTidb)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.InsertToFullStageTable)
			s.ExecuteServerlessDataWithRetryStep(s.CheckFullStageTableCount)
			s.StopChangefeedStep()
			s.WaitChangefeedStatusReadyStep(models.DataflowConnectorStatusPAUSED)
			s.EditChangefeedStep(s.buildEditTidbToServerlessTidbChangefeedReq())
			s.WaitChangefeedStatusReadyStep(models.DataflowConnectorStatusPAUSED)
			s.StartChangefeedStep()
			s.WaitChangefeedStatusReadyStep(models.DataflowConnectorStatusRUNNING)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.CreateIncreStageTable)
			s.ExecuteTidbDataWithDefaultClusterIDStep(s.InsertToIncreStageTable)
			s.ExecuteServerlessDataWithRetryStep(s.CheckTableIncreTestData)
			s.DeleteChangefeedStep()
		}))
}

func (s *CdcTidbToServerlessTiDBSceneTest) buildTestConnectToServerlessTidbReq() *models.DataflowConnectorCreateReq {
	dfPrivatelinkId, _ := strconv.ParseUint(s.DfPrivatelinkId, 10, 64)
	return &models.DataflowConnectorCreateReq{
		ConnectorType: models.DataflowConnectorTypeSERVERLESS_TIDB.Pointer(),
		ServerlessTidbInfo: &models.DataflowServerlessTidbInfo{
			ClusterId: &s.ServerlessClusterId,
			MysqlInfo: &models.DataflowMySQLInfo{
				Connection: &models.DataflowConnection{
					Username: s.ServerlessTidbUser,
					Password: s.Password,
				},
				PrivatelinkInfo: &models.DataflowPrivatelinkInfo{
					DfPrivatelinkEndpointId: dfPrivatelinkId,
				},
			},
		},
	}
}

func (s *CdcTidbToServerlessTiDBSceneTest) queryServerlessTidbInfo() {
	pathParam := map[string]string{
		"org_id":     s.DbaasRequestInfo.CommonEnv.OrgId,
		"project_id": s.DbaasRequestInfo.CommonEnv.ProjectId,
		"cluster_id": s.ServerlessClusterId,
	}
	callResp := s.DbaasRequestInfo.DbassRequest(s.T, "getClusterInfo", nil, pathParam, nil, nil)
	httpInfo := common.ParseHttpInfo(callResp)
	if httpInfo.RespStatusCode != common.StatusOK {
		errorMsg := fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
		l4g.Error(errorMsg)
		allure.FailNow(errorMsg)
	}

	clusterResp := &models.CentralGetClusterResp{}
	httpInfo.UnmarshalRespBody(clusterResp)
	s.ServerlessTidbPort = *clusterResp.Cluster.Endpoint.Port
	s.ServerlessTidbUser = *clusterResp.Cluster.Endpoint.Token + ".root"
}

func (s *CdcTidbToServerlessTiDBSceneTest) buildCreatePlEndpointReq() *models.DataflowCreatePrivateLinkEndpointReq {
	pathParam := map[string]string{
		"org_id":     s.DbaasRequestInfo.CommonEnv.OrgId,
		"project_id": s.DbaasRequestInfo.CommonEnv.ProjectId,
		"cluster_id": s.ServerlessClusterId,
	}
	callResp := s.DbaasRequestInfo.DbassRequest(s.T, "getEndpointService", nil, pathParam, nil, nil)
	httpInfo := common.ParseHttpInfo(callResp)
	l4g.Info("callResp:%s", callResp.JSON().Raw())
	if httpInfo.RespStatusCode != common.StatusOK {
		errorMsg := fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
		l4g.Error(errorMsg)
		allure.FailNow(errorMsg)
	}

	resp := &models.CentralGetPrivateLinkServiceResp{}
	httpInfo.UnmarshalRespBody(resp)
	return &models.DataflowCreatePrivateLinkEndpointReq{
		EndpointServiceName: resp.PrivatelinkService.Name,
	}
}

func (s *CdcTidbToServerlessTiDBSceneTest) buildCreateTidbToServerlessTidbChangefeedReq() *models.DataflowConnectorCreateReq {
	dfPrivatelinkId, _ := strconv.ParseUint(s.DfPrivatelinkId, 10, 64)
	return &models.DataflowConnectorCreateReq{
		ConnectorType: models.DataflowConnectorTypeSERVERLESS_TIDB.Pointer(),
		ServerlessTidbInfo: &models.DataflowServerlessTidbInfo{
			ClusterId: &s.ServerlessClusterId,
			MysqlInfo: &models.DataflowMySQLInfo{
				Connection: &models.DataflowConnection{
					Username: s.ServerlessTidbUser,
					Password: s.Password,
				},
				PrivatelinkInfo: &models.DataflowPrivatelinkInfo{
					DfPrivatelinkEndpointId: dfPrivatelinkId,
				},
				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(),
				},
			},
		},
		ChangefeedInstanceType: dpCommon.RCU2,
	}
}

func (s *CdcTidbToServerlessTiDBSceneTest) buildEditTidbToServerlessTidbChangefeedReq() *models.DataflowConnectorEditReq {
	dfPrivatelinkId, _ := strconv.ParseUint(s.DfPrivatelinkId, 10, 64)
	return &models.DataflowConnectorEditReq{
		ConnectorID:   s.ChangefeedId,
		ConnectorType: models.DataflowConnectorTypeSERVERLESS_TIDB.Pointer(),
		ServerlessTidbInfo: &models.DataflowServerlessTidbInfo{
			ClusterId: &s.ServerlessClusterId,
			MysqlInfo: &models.DataflowMySQLInfo{
				Connection: &models.DataflowConnection{
					Username: s.ServerlessTidbUser,
					Password: s.Password,
				},
				PrivatelinkInfo: &models.DataflowPrivatelinkInfo{
					DfPrivatelinkEndpointId: dfPrivatelinkId,
				},
				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(),
				},
			},
		},
	}
}