//go:build e2e
// +build e2e

package datalink

import (
	"fmt"
	"net/http"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/message"
	"github.com/tidbcloud/observability/pkg/common/structs"
	"github.com/tidbcloud/observability/test/ft/utils"
)

type CreateCase struct {
	name string
	message.CreateClusterReq
	expectedCode  int
	contentAssert func(t *testing.T, resp *structs.ObservabilityCluster)
}

func postAlertManager(alertName, summary string) {
	conf := utils.GetDataLinkConfig()
	utils.PostAlert(conf.AlertEndPoint, []utils.AlertRequest{
		{
			Labels: map[string]string{
				"severity":  "warning",
				"component": "observability-regional-server",
				"project":   conf.AlertType,
				"alertname": alertName,
				"buketname": conf.BucketName,
			},
			Annotations: map[string]string{
				"summary": summary,
			},
			StartsAt: time.Now().Format("2006-01-02T15:04:05Z07:00"),
		},
	})
}

func TestPrometheusWrite(t *testing.T) {
	cases := []CreateCase{
		{
			name: "prometheus-write-test",
			CreateClusterReq: utils.HandleWithDefault(func(req message.CreateClusterReq) message.CreateClusterReq {
				req.ApplicationID = "application-write"
				req.TenantID = "write-test"
				return req
			}),
			expectedCode: 202,
			contentAssert: func(t *testing.T, cluster *structs.ObservabilityCluster) {
				assert.Equal(t, "application-write", cluster.ApplicationID)
			},
		},
	}
	var cluster *structs.ObservabilityCluster
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			// create cluster
			request := utils.CreateReq(c.CreateClusterReq)
			resp, err := http.DefaultClient.Do(request)
			assert.NoError(t, err)
			defer resp.Body.Close()
			assert.Equal(t, c.expectedCode, resp.StatusCode)
			clustersResp := &message.CreateClusterResp{}
			utils.HandleResp(t, resp, clustersResp)
			assert.NotEqual(t, 0, len(clustersResp.ID))

			var status string
			nowTime := time.Now()
			createClusterStuatus := true
			for status != string(constants.ACTIVE) {
				if time.Since(nowTime).Seconds() > 180 {
					createClusterStuatus = false
					break
				}
				res := utils.GetCluster(clustersResp.ID)
				status = string(res.Status)
				time.Sleep(20 * time.Second)
			}
			assert.Equal(t, true, createClusterStuatus)
			finalCluster := utils.GetCluster(clustersResp.ID)
			cluster = finalCluster
			fmt.Printf("create cluster %s for prometheus write, endpoint service name: %s\n", clustersResp.ID, cluster.MonitorInfo.EndpointServiceName)
		})
	}

	utils.CreateEndPoint(cluster.ID, cluster.MonitorInfo.EndpointServiceName)

	t.Run("prometheus write", func(t *testing.T) {
		assert.NotEmpty(t, cluster.ID)
		dns := utils.GetEndPointDNS()
		assert.NotEmpty(t, dns)
		dataLinkConfig := utils.GetDataLinkConfig()
		utils.StartPrometheusWrite(utils.StartWriteParam{
			ClusterID: cluster.ID,
			NameSpace: dataLinkConfig.AgentNameSpace,
			Name:      dataLinkConfig.AgentName,
			DNS:       dns,
		})
		fmt.Println("start prometheus write, clusterID:", cluster.ID, "tentantID:", cluster.TenantID)
		defer fmt.Println("end prometheus write, clusterID:", cluster.ID, "tentantID:", cluster.TenantID)
		time.Sleep(dataLinkConfig.PrometheusWriteDuration)
		utils.StopAgentWrite(dataLinkConfig.AgentNameSpace, dataLinkConfig.AgentName)

		promWriteRes := utils.GetTSDBStatus(cluster.ID)
		if (promWriteRes.Status != "success" || promWriteRes.Data.HeadStats.NumSeries == 0) &&
			dataLinkConfig.Env == "pod" {
			postAlertManager("TestPrometheusWriteFailed", "promWriteRes.Data.HeadStats.NumSeries = 0")
		}
		assert.Equal(t, "success", promWriteRes.Status)
		assert.NotZero(t, promWriteRes.Data.HeadStats.NumSeries)

		assert.Equal(t, 10, len(promWriteRes.Data.SeriesCountByMetricName))
		metricKV := make(map[string]int)
		for _, metric := range promWriteRes.Data.SeriesCountByMetricName {
			metricKV[metric.Name] = metric.Value
		}

		metricsStatus := metricsCheck(metricKV)

		if !metricsStatus && dataLinkConfig.Env == "pod" {
			postAlertManager("TestPrometheusWriteMetricsCheckFailed", "metricsCheck failed")
		}
		assert.Equal(t, true, metricsStatus)

	})

	utils.DeleteCluster(cluster.ID)
}

func metricsCheck(checkMetrics map[string]int) bool {
	metrics := []string{
		"demo_bytes_histogram_bucket", "demo_bytes_summary", "demo_bytes_summary_count", "demo_bytes_summary_avg",
		"demo_status_gauge", "demo_bytes_histogram_sum", "demo_bytes_histogram_count",
	}
	for _, metric := range metrics {
		if v, ok := checkMetrics[metric]; !ok {
			return false
		} else {
			fmt.Println("metricsCheck, metric_name:", metric, "metric_value:", v)
			if v <= 0 {
				return false
			}
		}
	}
	return true
}

func TestLogWrite(t *testing.T) {
	dataLinkConfig := utils.GetDataLinkConfig()
	cases := []CreateCase{
		{
			name: "log-write-test",
			CreateClusterReq: utils.HandleWithDefault(func(req message.CreateClusterReq) message.CreateClusterReq {
				req.ApplicationID = "application-write"
				req.TenantID = "log-write-test"
				req.AgentAccessARN = dataLinkConfig.AgentARN
				return req
			}),
			expectedCode: 202,
			contentAssert: func(t *testing.T, cluster *structs.ObservabilityCluster) {
				assert.Equal(t, "application-write", cluster.ApplicationID)
			},
		},
	}
	var cluster *structs.ObservabilityCluster
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			request := utils.CreateReq(c.CreateClusterReq)
			resp, err := http.DefaultClient.Do(request)
			assert.NoError(t, err)
			defer resp.Body.Close()
			assert.Equal(t, c.expectedCode, resp.StatusCode)
			fmt.Println("create cluster response: ", resp)

			clustersResp := &message.CreateClusterResp{}
			utils.HandleResp(t, resp, clustersResp)
			assert.NotEqual(t, 0, len(clustersResp.ID))
			fmt.Println("create cluster response ID: ", clustersResp.ID)
			var status string
			nowTime := time.Now()
			createSuccess := true
			for status != string(constants.ACTIVE) {
				if time.Since(nowTime).Seconds() > 180 {
					createSuccess = false
					break
				}
				res := utils.GetCluster(clustersResp.ID)
				status = string(res.Status)
				time.Sleep(20 * time.Second)
			}
			assert.True(t, createSuccess)
			finalCluster := utils.GetCluster(clustersResp.ID)
			c.contentAssert(t, finalCluster)
			cluster = finalCluster
		})
	}

	t.Run("log write", func(t *testing.T) {
		assert.NotEmpty(t, cluster.ID)
		assert.True(t, len(cluster.StorageInfo) > 0)
		bucket := dataLinkConfig.BucketName
		for _, storage := range cluster.StorageInfo {
			if storage.Category == "logs" {
				bucket = storage.BucketName
				break
			}
		}
		utils.SetDataLinkBuket(bucket)

		utils.StartLogWrite(utils.StartWriteParam{
			ClusterID:  cluster.ID,
			Buket:      bucket,
			NameSpace:  dataLinkConfig.AgentNameSpace,
			Name:       dataLinkConfig.AgentName,
			TenantID:   cluster.TenantID,
			AssumeRole: cluster.AgentInfo.RoleARN,
		})
		fmt.Println("start log write, clusterID:", cluster.ID, "tentantID:", cluster.TenantID, "Bucket:", bucket)
		time.Sleep(dataLinkConfig.LogWriteDuration)
		utils.StopAgentWrite(dataLinkConfig.AgentNameSpace, dataLinkConfig.AgentName)
		time.Sleep(600 * time.Second)
		logFlieSizes := utils.LogCheckAndDelete(cluster.TenantID)
		if len(logFlieSizes) == 0 && dataLinkConfig.Env == "pod" {
			alertName := "gcsWriteFailed"
			if cluster.Vendor == constants.AWS {
				alertName = "S3WriteFailed"
			}
			fmt.Println(alertName)
			postAlertManager(alertName, "logFlieSizes equal zero")
		}
		assert.NotZero(t, len(logFlieSizes))
		for logFile, size := range logFlieSizes {
			fmt.Println("log-name:", fmt.Sprintf(dataLinkConfig.LogKeyPrefix, cluster.TenantID)+"/"+logFile,
				"size:", size)
			assert.NotZero(t, size)
		}
	})
	fmt.Println("end log write, clusterID:", cluster.ID, "tentantID:", cluster.TenantID)
	utils.DeleteCluster(cluster.ID)
}