package cdc

import (
	"fmt"
	"github.com/avast/retry-go"
	"github.com/pingcap/log"
	"strings"
	"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-gourd/handler"
)

func (s *CommonSceneTest) GetAccountId() {
	allure.Step(
		allure.Description(fmt.Sprintf("get account id")),
		allure.Action(func() {
			pathParam := map[string]string{
				"org_id":     s.DbaasRequestInfo.CommonEnv.OrgId,
				"project_id": s.DbaasRequestInfo.CommonEnv.ProjectId,
				"cluster_id": s.ClusterId,
			}

			err := retry.Do(
				func() error {
					callResp := s.DbaasRequestInfo.DbassRequest(s.T, "adminProfile", nil, pathParam, nil, nil)
					httpInfo := common.ParseHttpInfo(callResp)

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
					}
					s.AccountId = callResp.JSON().Path("$.gcp_account_id").String().Raw()
					return nil
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*10),
				retry.Attempts(6),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}
		}),
	)
}

func (s *CommonSceneTest) GetClusterVersion() {
	allure.Step(
		allure.Description(fmt.Sprintf("get cluster version")),
		allure.Action(func() {
			pathParam := map[string]string{
				"org_id":     s.DbaasRequestInfo.CommonEnv.OrgId,
				"project_id": s.DbaasRequestInfo.CommonEnv.ProjectId,
				"cluster_id": s.ClusterId,
			}

			err := retry.Do(
				func() error {
					callResp := s.DbaasRequestInfo.DbassRequest(s.T, "getClusterInfo", nil, pathParam, nil, nil)
					httpInfo := common.ParseHttpInfo(callResp)

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
					}
					s.Version = callResp.JSON().Path("$.cluster.version").String().Raw()
					return nil
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*10),
				retry.Attempts(6),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}
		}),
	)
}

func (s *CommonSceneTest) CreateChangefeedStep(req *models.DataflowConnectorCreateReq) {
	allure.Step(
		allure.Description(fmt.Sprintf("create %s changefeed step", string(*req.ConnectorType))),
		allure.Action(func() {
			pathParam := map[string]string{
				"org_id":     s.DbaasRequestInfo.CommonEnv.OrgId,
				"project_id": s.DbaasRequestInfo.CommonEnv.ProjectId,
				"cluster_id": s.ClusterId,
			}
			err := retry.Do(
				func() error {
					callResp := s.DbaasRequestInfo.DbassRequest(s.T, "connectorConfirm", req, pathParam, nil, nil)
					httpInfo := common.ParseHttpInfo(callResp, common.ReqBodyOp(req))

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
					}

					createChangefeedResp := &models.DataflowConnectorID{}
					httpInfo.UnmarshalRespBody(createChangefeedResp)

					if createChangefeedResp.ConnectorID == nil {
						return fmt.Errorf("changefeed id is empty, http details is: %s", httpInfo.ToString())
					}
					s.ChangefeedId = *createChangefeedResp.ConnectorID
					return nil
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*10),
				retry.Attempts(6),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}
		}),
	)
}

func (s *CommonSceneTest) WaitChangefeedStatusReadyStep(expectStatus models.DataflowConnectorStatus) {
	allure.Step(
		allure.Description(fmt.Sprintf("check changefeed %s is %s step", s.ChangefeedId, expectStatus)),
		allure.Action(func() {
			pathParam := map[string]string{
				"org_id":       s.DbaasRequestInfo.CommonEnv.OrgId,
				"project_id":   s.DbaasRequestInfo.CommonEnv.ProjectId,
				"cluster_id":   s.ClusterId,
				"connector_id": s.ChangefeedId,
			}
			err := retry.Do(
				func() error {
					callResp := s.DbaasRequestInfo.DbassRequest(s.T, "getConnectorInfo", nil, pathParam, nil, nil)
					httpInfo := common.ParseHttpInfo(callResp)

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
					}

					getChangefeedInfoResp := &models.DataflowConnector{}
					httpInfo.UnmarshalRespBody(getChangefeedInfoResp)
					if *getChangefeedInfoResp.Status != expectStatus {
						return fmt.Errorf("coonnector id is is: %s, expect status is %s, actural status is %s, http details is: %s",
							s.ChangefeedId, expectStatus, *getChangefeedInfoResp.Status, httpInfo.ToString())
					}
					return nil
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*10),
				retry.Attempts(6*50),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}
		}),
	)
}

func (s *CommonSceneTest) CheckChangefeedRcuInstanceTypeStep(expectRcuInstanceType string) {
	allure.Step(
		allure.Description(fmt.Sprintf("check changefeed rcu instance type %s step", expectRcuInstanceType)),
		allure.Action(func() {
			pathParam := map[string]string{
				"org_id":       s.DbaasRequestInfo.CommonEnv.OrgId,
				"project_id":   s.DbaasRequestInfo.CommonEnv.ProjectId,
				"cluster_id":   s.ClusterId,
				"connector_id": s.ChangefeedId,
			}
			err := retry.Do(
				func() error {
					callResp := s.DbaasRequestInfo.DbassRequest(s.T, "getConnectorInfo", nil, pathParam, nil, nil)
					httpInfo := common.ParseHttpInfo(callResp)

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
					}

					getChangefeedInfoResp := &models.DataflowConnector{}
					httpInfo.UnmarshalRespBody(getChangefeedInfoResp)
					if getChangefeedInfoResp.ChangefeedInstanceType != expectRcuInstanceType {
						return fmt.Errorf("coonnector id is is: %s, expect rcu instance type is %s, actural rcu instance type is %s, http details is: %s",
							s.ChangefeedId, expectRcuInstanceType, getChangefeedInfoResp.ChangefeedInstanceType, httpInfo.ToString())
					}
					return nil
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*10),
				retry.Attempts(6*15),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}
		}),
	)
}

func (s *CommonSceneTest) StopChangefeedStep() {
	allure.Step(
		allure.Description(fmt.Sprintf("stop changefeed %s step", s.ChangefeedId)),
		allure.Action(func() {
			pathParam := map[string]string{
				"org_id":       s.DbaasRequestInfo.CommonEnv.OrgId,
				"project_id":   s.DbaasRequestInfo.CommonEnv.ProjectId,
				"cluster_id":   s.ClusterId,
				"connector_id": s.ChangefeedId,
			}

			err := retry.Do(
				func() error {
					callResp := s.DbaasRequestInfo.DbassRequest(s.T, "stopConnector", nil, pathParam, nil, nil)
					httpInfo := common.ParseHttpInfo(callResp)

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
					}
					return nil
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*10),
				retry.Attempts(6),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}
		}),
	)
}

func (s *CommonSceneTest) EditChangefeedStep(callReq *models.DataflowConnectorEditReq) {
	allure.Step(
		allure.Description(fmt.Sprintf("edit changefeed %s step", s.ChangefeedId)),
		allure.Action(func() {
			pathParam := map[string]string{
				"org_id":     s.DbaasRequestInfo.CommonEnv.OrgId,
				"project_id": s.DbaasRequestInfo.CommonEnv.ProjectId,
				"cluster_id": s.ClusterId,
			}

			err := retry.Do(
				func() error {
					callResp := s.DbaasRequestInfo.DbassRequest(s.T, "editConnector", callReq, pathParam, nil, nil)
					httpInfo := common.ParseHttpInfo(callResp, common.ReqBodyOp(callReq))

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
					}
					return nil
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*10),
				retry.Attempts(12),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}

		}),
	)
}

func (s *CommonSceneTest) ScaleChangefeedStep(callReq *models.DataflowConnectorScaleReq) {
	allure.Step(
		allure.Description("scale changefeed step"),
		allure.Action(func() {
			pathParam := map[string]string{
				"org_id":        s.DbaasRequestInfo.CommonEnv.OrgId,
				"project_id":    s.DbaasRequestInfo.CommonEnv.ProjectId,
				"cluster_id":    s.ClusterId,
				"changefeed_id": s.ChangefeedId,
			}

			err := retry.Do(
				func() error {
					callResp := s.DbaasRequestInfo.DbassRequest(s.T, "scaleChangefeed", callReq, pathParam, nil, nil)
					httpInfo := common.ParseHttpInfo(callResp, common.ReqBodyOp(callReq))

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
					}
					return nil
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*30),
				retry.Attempts(20),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}

		}),
	)
}

func (s *CommonSceneTest) StartChangefeedStep() {
	allure.Step(
		allure.Description(fmt.Sprintf("start changefeed %s step", s.ChangefeedId)),
		allure.Action(func() {
			pathParam := map[string]string{
				"org_id":       s.DbaasRequestInfo.CommonEnv.OrgId,
				"project_id":   s.DbaasRequestInfo.CommonEnv.ProjectId,
				"cluster_id":   s.ClusterId,
				"connector_id": s.ChangefeedId,
			}
			err := retry.Do(
				func() error {
					callResp := s.DbaasRequestInfo.DbassRequest(s.T, "startConnector", nil, pathParam, nil, nil)
					httpInfo := common.ParseHttpInfo(callResp)

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
					}
					return nil
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*10),
				retry.Attempts(6),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}
		}),
	)
}

func (s *CommonSceneTest) DeleteChangefeedStep() {
	allure.Step(
		allure.Description(fmt.Sprintf("delete changefeed %s step", s.ChangefeedId)),
		allure.Action(func() {
			log.Info("delete changefeed step ")
			pathParam := map[string]string{
				"org_id":       s.DbaasRequestInfo.CommonEnv.OrgId,
				"project_id":   s.DbaasRequestInfo.CommonEnv.ProjectId,
				"cluster_id":   s.ClusterId,
				"connector_id": s.ChangefeedId,
			}

			err := retry.Do(
				func() error {
					callResp := s.DbaasRequestInfo.DbassRequest(s.T, "deleteConnector", nil, pathParam, nil, nil)
					httpInfo := common.ParseHttpInfo(callResp)

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
					}
					return nil
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*10),
				retry.Attempts(6),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}
		}),
	)
}

func (s *CommonSceneTest) TestDownStreamConnectionStep(req *models.DataflowConnectorCreateReq) {
	allure.Step(
		allure.Description(fmt.Sprintf("test %s connection  step", string(*req.ConnectorType))),
		allure.Action(func() {
			pathParam := map[string]string{
				"org_id":     s.DbaasRequestInfo.CommonEnv.OrgId,
				"project_id": s.DbaasRequestInfo.CommonEnv.ProjectId,
				"cluster_id": s.ClusterId,
			}

			err := retry.Do(
				func() error {
					callResp := s.DbaasRequestInfo.DbassRequest(s.T, "connectorTest", req, pathParam, nil, nil)
					httpInfo := common.ParseHttpInfo(callResp, common.ReqBodyOp(req))

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
					}
					return nil
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*10),
				retry.Attempts(6),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}
		}),
	)
}

func (s *CommonSceneTest) CreatePrivatelinkEndpoint(req *models.DataflowCreatePrivateLinkEndpointReq) {
	allure.Step(
		allure.Description("create privatelink endpoint step"),
		allure.Action(func() {
			pathParam := map[string]string{
				"org_id":     s.DbaasRequestInfo.CommonEnv.OrgId,
				"project_id": s.DbaasRequestInfo.CommonEnv.ProjectId,
				"cluster_id": s.ClusterId,
			}

			err := retry.Do(
				func() error {
					callResp := s.DbaasRequestInfo.DbassRequest(s.T, "createDfPlEndpoint", req, pathParam, nil, nil)
					httpInfo := common.ParseHttpInfo(callResp)

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("connector id is is: %s,  status code is: %d, http details is: %s", s.ChangefeedId, httpInfo.RespStatusCode, httpInfo.ToString())
					}

					resp := &models.DataflowCreatePrivateLinkEndpointResp{}
					httpInfo.UnmarshalRespBody(resp)
					if resp == nil || resp.Id == "" {
						return fmt.Errorf("create privatelink endepoint resp is empty, http details is: %s", httpInfo.ToString())
					}
					s.DfPrivatelinkId = resp.Id
					return nil
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*10),
				retry.Attempts(6),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}
		}),
	)
}

func (s *CommonSceneTest) WaitPrivatelinkEndpointReadyStep() {
	allure.Step(
		allure.Description(fmt.Sprintf("wait private link endpoint %s ready step", s.DfPrivatelinkId)),
		allure.Action(func() {
			pathParam := map[string]string{
				"org_id":     s.DbaasRequestInfo.CommonEnv.OrgId,
				"project_id": s.DbaasRequestInfo.CommonEnv.ProjectId,
				"cluster_id": s.ClusterId,
				"id":         s.DfPrivatelinkId,
			}
			err := retry.Do(
				func() error {
					callResp := s.DbaasRequestInfo.DbassRequest(s.T, "getDfPlEndpoint", nil, pathParam, nil, nil)
					httpInfo := common.ParseHttpInfo(callResp)

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
					}

					resp := &models.DataflowGetPrivateLinkEndpointResp{}
					httpInfo.UnmarshalRespBody(resp)
					if resp.Phase != models.DataflowGetPrivateLinkEndpointResp_Available {
						return fmt.Errorf("connector id is is: %s, df privatelink id is: %s, expect status is %s, actural status is %s, http details is: %s",
							s.ChangefeedId, s.DfPrivatelinkId, models.DataflowGetPrivateLinkEndpointResp_Available, resp.Phase, httpInfo.ToString())
					}
					return nil
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*10),
				retry.Attempts(6*8),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}
		}),
	)
}

func (s *CommonSceneTest) GetMetrics() {
	allure.Step(
		allure.Description(fmt.Sprintf("get metrics")),
		allure.Action(func() {
			time.Sleep(time.Minute * 2)
			var env string
			if handler.EnvFile == "staging" || handler.EnvFile == "dev" {
				env = fmt.Sprintf("%s-", handler.EnvFile)
			} else {
				env = ""
			}
			pathParam := map[string]string{
				"env":        env,
				"org_id":     s.DbaasRequestInfo.CommonEnv.OrgId,
				"project_id": s.DbaasRequestInfo.CommonEnv.ProjectId,
				"cluster_id": s.ClusterId,
			}
			var replicaRowsPerSec string

			if strings.Compare("v6.3", s.Version) > 0 {
				replicaRowsPerSec = "ticdc_sink_total_flushed_rows_count"
			} else if strings.Compare("v7.0", s.Version) > 0 {
				replicaRowsPerSec = "ticdc_sinkv2_batch_row_count_sum"
			} else {
				replicaRowsPerSec = "ticdc_sink_batch_row_count_sum"
			}

			endTime := time.Now().Unix()
			startTime := time.Now().Add(-2 * time.Minute).Unix()

			queryCheckpointTsParam := map[string]string{
				"start": fmt.Sprintf("%d", startTime),
				"end":   fmt.Sprintf("%d", endTime),
				"step":  "30",
				"query": fmt.Sprintf("max(ticdc_owner_checkpoint_ts_lag{changefeed=\"changefeed-%s\"}) by (changefeed)", s.ChangefeedId),
			}

			queryReplicaRowParam := map[string]string{
				"start": fmt.Sprintf("%d", startTime),
				"end":   fmt.Sprintf("%d", endTime),
				"step":  "30",
				"query": fmt.Sprintf("sum(rate(%s{changefeed=\"changefeed-%s\"}[1m])) by (changefeed)", replicaRowsPerSec, s.ChangefeedId),
			}

			err := retry.Do(
				func() error {
					callResp := s.DbaasRequestInfo.DbassRequest(s.T, "getChangefeedMetric", nil, pathParam, queryCheckpointTsParam, nil)
					httpInfo := common.ParseHttpInfo(callResp)

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
					}
					resp := &common.Metrics{}
					httpInfo.UnmarshalRespBody(resp)
					if len(resp.Data.Result) == 0 {
						return fmt.Errorf("cannot get checkpointts metrics, resp:%v", callResp.JSON().Raw())
					}
					callResp = s.DbaasRequestInfo.DbassRequest(s.T, "getChangefeedMetric", nil, pathParam, queryReplicaRowParam, nil)
					httpInfo = common.ParseHttpInfo(callResp)

					if httpInfo.RespStatusCode != common.StatusOK {
						return fmt.Errorf("status code is: %d, http details is: %s", httpInfo.RespStatusCode, httpInfo.ToString())
					}
					resp = &common.Metrics{}
					httpInfo.UnmarshalRespBody(resp)
					if len(resp.Data.Result) == 0 {
						return fmt.Errorf("cannot get replicaRowsPerSec metrics, resp:%v", callResp.JSON().Raw())
					}
					l4g.Info("replicaRowsPerSec resp:%s, queryReplicaRowParam:%s", callResp.JSON().Raw(), queryReplicaRowParam)
					return nil
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*10),
				retry.Attempts(6),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}
		}),
	)
}