package slprober

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"cloud.google.com/go/storage"
	"github.com/ajg/form"
	v2Aws "github.com/aws/aws-sdk-go-v2/aws"
	v2Config "github.com/aws/aws-sdk-go-v2/config"
	v2S3 "github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/gogo/protobuf/proto"
	"github.com/golang/snappy"
	"google.golang.org/api/iterator"
	"gopkg.in/yaml.v2"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	ctrl "sigs.k8s.io/controller-runtime"
)

type RegionConfig struct {
	Config       Config
	Env          string
	Region       string
	Vendor       string
	RegionDomain string
	UserArn      string
}

type Config struct {
	Auth0        Auth0Req     `yaml:"auth0"`
	ClinicServer ClinicServer `yaml:"clinicServer"`
	ServerURL    string       `yaml:"serverURL"`
	BizType      []string     `yaml:"bizType"`
	OldPath      []string     `yaml:"oldPath"`
}

type Auth0Req struct {
	GrantType    string `yaml:"grantType" json:"grant_type" form:"grant_type"`
	ClientID     string `yaml:"clientID" json:"client_id" form:"client_id"`
	ClientSecret string `yaml:"clientSecret" json:"client_secret" form:"client_secret"`
	Audience     string `yaml:"audience" json:"audience" form:"audience"`
}

type Auth0Resp struct {
	AccessToken string `json:"access_token" form:"access_token"`
}

type ClinicServer struct {
	ClinicToken        string `yaml:"clinicToken"`
	ClinicTokenStaging string `yaml:"clinicTokenStaging"`
	ClinicURL          string `yaml:"clinicURL"`
}

type CommonReqInfo struct {
	Vendor string `json:"vendor" form:"vendor" validate:"required,max=16"`
	Region string `json:"region" form:"region" validate:"required,max=16"`
}

type ClusterBaseInfo struct {
	CommonReqInfo
	TenantID           string   `json:"tenantID" form:"tenantID" validate:"required,max=20"`
	ProjectID          string   `json:"projectID" form:"projectID" validate:"required,max=20"`
	ApplicationID      string   `json:"applicationID" form:"applicationID" validate:"required,max=20"`
	BizType            string   `json:"bizType" form:"bizType" validate:"required,max=32"`
	ID                 string   `json:"id" form:"id" validate:"max=64"`
	Tags               []string `json:"tags" form:"tags"`
	Version            string   `json:"version" form:"version"`
	ShareMode          string   `json:"shareMode" form:"shareMode" validate:"required,oneof=exclusive shared virtual"`
	Note               string   `json:"note" form:"note"`
	DataPropertyRights int      `json:"dataPropertyRights" form:"dataPropertyRights" validate:"oneof=0 1"`
}

var regionConfig *RegionConfig
var token string
var ApiPrefix = "/api/v1"
var kubeClient *kubernetes.Clientset

func GetConfig() *RegionConfig {
	if regionConfig == nil {
		panic("config is nil")
	}
	return regionConfig
}
func DefaultAuth0Req() Auth0Req {
	return regionConfig.Config.Auth0
}
func PrepareToken() {
	body, err := form.EncodeToString(DefaultAuth0Req())
	if err != nil {
		panic(err)
	}
	resp, err := http.Post("https://tidb-soc2.us.auth0.com/oauth/token",
		"application/x-www-form-urlencoded",
		strings.NewReader(body))
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()
	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		panic(err)
	}
	auth0Resp := &Auth0Resp{}
	err = json.Unmarshal(respContent, auth0Resp)
	if err != nil {
		panic(err)
	}
	token = auth0Resp.AccessToken
}

func PrepareConfigInfo(path string) {
	in, err := os.ReadFile(path)
	if err != nil {
		panic(err)
	}
	var c Config
	err = yaml.Unmarshal(in, &c)
	if err != nil {
		panic(err)
	}
	rc := RegionConfig{
		Config:       c,
		Env:          os.Getenv("ENV"),
		Region:       os.Getenv("REGION"),
		Vendor:       os.Getenv("VENDOR"),
		RegionDomain: os.Getenv("REGION_DOMAIN"),
		UserArn:      os.Getenv("USER_ARN"),
	}

	regionConfig = &rc
}

func PrepareKubeClient() {
	var err error
	os.Setenv("KUBECONFIG", fmt.Sprintf("%s/.kube/config", os.Getenv("HOME")))
	kubeConfig := ctrl.GetConfigOrDie()
	kubeClient, err = kubernetes.NewForConfig(kubeConfig)
	if err != nil {
		panic(err)
	}
}

type QueryClustersReq struct {
	CommonReqInfo
	TenantID      string `json:"tenantID" form:"tenantID"`
	ProjectID     string `json:"projectID" form:"projectID"`
	ApplicationID string `json:"applicationID" form:"applicationID"`
	BizType       string `json:"bizType" form:"bizType"`
}

type QueryClustersResp struct {
	Clusters []ClusterBaseInfo `json:"clusters"`
}

type DetailClusterReq struct {
	CommonReqInfo
	ID string `json:"id" swaggerignore:"true"`
}

type DetailClusterResp struct {
	ObservabilityCluster
}

type CreateClusterReq struct {
	ClusterBaseInfo
	AgentAccessARN    string `json:"agentAccessARN" form:"agentAccessARN" validate:"required"`
	AgentManagerARN   string `json:"agentManagerARN" form:"agentManagerARN"`
	AgentTemplateName string `json:"agentTemplateName" form:"agentTemplateName"`
	MonitorSpecID     string `json:"monitorSpecID" form:"monitorSpecID"`
	HostClusterID     string `json:"hostClusterID" form:"hostClusterID"`
	//MonitorPromRules  prometheusv1.PrometheusRuleSpec `json:"monitorPromRules" form:"monitorPromRules"`
	RuleTemplateID string `json:"ruleTemplateID" form:"ruleTemplateID"`
	MuteAlert      bool   `json:"muteAlert" form:"muteAlert"`
}

type CreateClusterResp struct {
	ObservabilityCluster
}

type ObservabilityCluster struct {
	ClusterBaseInfo
	StorageInfo   []StorageInfo     `json:"storageInfo" form:"storageInfo"`
	MonitorInfo   MonitorInfo       `json:"monitorInfo" form:"monitorInfo"`
	AgentInfo     AgentInfo         `json:"agentInfo" form:"agentInfo"`
	AlertInfo     AlertInfo         `json:"alertInfo" form:"alertInfo"`
	Status        string            `json:"status" form:"status"`
	LastScaleTime map[string]string `json:"lastScaleTime" form:"lastScaleTime"`
}

type StorageInfo struct {
	BucketName string `json:"bucketName" form:"bucketName"`
	Category   string `json:"category" form:"category"`
	DataARN    string `json:"dataARN" form:"dataARN"`
}

type MonitorInfo struct {
	SpecID string `json:"specID" form:"specID"`
	//Storage             string `json:"storage" form:"storage"`
	EndpointServiceName string `json:"endpointServiceName" form:"endpointServiceName"`
	DNSName             string `json:"dnsName" form:"dnsName"`
}

type AlertInfo struct {
	MuteAlert      bool   `json:"muteAlert" form:"muteAlert"`
	RuleTemplateID string `json:"ruleTemplateID" form:"ruleTemplateID"`
}

type AgentInfo struct {
	RoleARN string `json:"roleARN" form:"roleARN"`
}

type QueryResponse struct {
	Status    string `json:"status"`
	IsPartial bool   `json:"isPartial"`
	Data      Data   `json:"data"`
}

type Data struct {
	ResultType string   `json:"resultType"`
	Result     []Result `json:"result"`
}

type Result struct {
	Metric map[string]string `json:"metric"`
	Value  []interface{}     `json:"value"`
}

func QueryCluster(queryReq QueryClustersReq) *QueryClustersResp {
	request := QueryReq(queryReq)
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()
	clustersResp := &QueryClustersResp{}
	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		panic(err)
	}
	err = UnmarshalRespData(respContent, clustersResp)
	if err != nil {
		panic(err)
	}
	return clustersResp
}

func QueryReq(req QueryClustersReq) *http.Request {
	c := GetConfig()
	queryString := fmt.Sprintf("%s%s/clusters/?vendor=%s&region=%s&bizType=%s",
		c.Config.ServerURL, ApiPrefix, c.Vendor, c.Region, req.BizType)
	request, _ := http.NewRequest("GET", queryString, nil)
	WithHeader(request)
	return request
}

func GetClusterDetail(req DetailClusterReq) *ObservabilityCluster {
	request := DetailReq(req)
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusNotFound {
		return nil
	}
	clustersResp := &DetailClusterResp{}
	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		panic(err)
	}

	err = UnmarshalRespData(respContent, clustersResp)
	if err != nil {
		panic(err)
	}
	return &clustersResp.ObservabilityCluster
}

func DetailReq(req DetailClusterReq) *http.Request {
	c := GetConfig()
	api := fmt.Sprintf("%s%s/clusters/%s?vendor=%s&region=%s",
		c.Config.ServerURL, ApiPrefix, req.ID, req.Vendor, req.Region)
	body, _ := json.Marshal(req)
	request, _ := http.NewRequest("GET", api, strings.NewReader(string(body)))
	WithHeader(request)
	return request
}

func CreateCluster(req CreateClusterReq) (*http.Response, error) {
	request := CreateReq(req)
	resp, err := http.DefaultClient.Do(request)
	return resp, err
}

func CreateReq(req CreateClusterReq) *http.Request {
	api := fmt.Sprintf("%s%s/clusters/", GetConfig().Config.ServerURL, ApiPrefix)

	body, _ := json.Marshal(req)
	request, _ := http.NewRequest("POST", api, strings.NewReader(string(body)))
	WithHeader(request)
	return request
}

func GetCluster(id string) (*http.Response, error) {
	c := GetConfig()
	req := DetailClusterReq{
		CommonReqInfo: CommonReqInfo{
			Vendor: c.Vendor,
			Region: c.Region,
		},
		ID: id,
	}
	request := GetReq(req)
	resp, err := http.DefaultClient.Do(request)
	return resp, err
}

func GetReq(req DetailClusterReq) *http.Request {
	api := fmt.Sprintf("%s%s/clusters/%s?vendor=%s&region=%s",
		GetConfig().Config.ServerURL, ApiPrefix, req.ID, req.Vendor, req.Region)
	body, _ := json.Marshal(req)
	request, _ := http.NewRequest("GET", api, strings.NewReader(string(body)))
	WithHeader(request)
	return request
}

func UnmarshalRespData(data []byte, v interface{}) error {
	type Result struct {
		Code int         `json:"code"`
		Data interface{} `json:"data"`
	}
	result := &Result{
		Data: v,
	}
	return json.Unmarshal(data, result)
}

func WithHeader(request *http.Request) {
	request.Header.Set("Content-Type", "application/json")
	request.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token))
}

func DefaultSession(region string) (v2Aws.Config, error) {
	cfg, err := v2Config.LoadDefaultConfig(context.TODO(),
		v2Config.WithRegion(region))
	if err != nil {
		return v2Aws.Config{}, fmt.Errorf("Failed to create session, %v", err)
	}

	return cfg, err
}

func GetMinutesSinceLastUpdateTimeS3(cluster *ObservabilityCluster, cfg v2Aws.Config, createTime time.Time) (lastModifiedTime time.Time,
	minutesToLastModified int, curTime time.Time, err error) {
	prefix := fmt.Sprintf("%d/%s/%s/%s/logs/", cluster.ClusterBaseInfo.DataPropertyRights, cluster.TenantID, cluster.ProjectID, cluster.ApplicationID)
	objects, err := v2S3.NewFromConfig(cfg).ListObjects(context.TODO(), &v2S3.ListObjectsInput{
		Bucket:    v2Aws.String(cluster.StorageInfo[0].BucketName),
		Prefix:    v2Aws.String(prefix),
		Delimiter: v2Aws.String("/"),
	})
	if err != nil {
		return
	}

	subPrefixes := make([]string, 0)
	for _, commonPrefix := range objects.CommonPrefixes {
		subPrefixFullPathArray := strings.Split(*commonPrefix.Prefix, "/")
		subPrefixes = append(subPrefixes, subPrefixFullPathArray[len(subPrefixFullPathArray)-2])
	}

	lastModifiedTime = createTime
	minutesToLastModified = int(time.Since(lastModifiedTime).Minutes())
	for _, s := range subPrefixes {
		objects, err = v2S3.NewFromConfig(cfg).ListObjects(context.TODO(), &v2S3.ListObjectsInput{
			Bucket: v2Aws.String(cluster.StorageInfo[0].BucketName),
			Prefix: v2Aws.String(fmt.Sprintf("%s%s/", prefix, s)),
		})
		if err != nil {
			return
		}
		tempCurTime := time.Now()
		// get a fundamental minutesToLastModified
		if len(objects.Contents) > 0 {
			o := objects.Contents[len(objects.Contents)-1]
			tempMinutesToLastModified := int(tempCurTime.Sub(*o.LastModified).Minutes())
			if tempMinutesToLastModified < minutesToLastModified {
				lastModifiedTime = *o.LastModified
				minutesToLastModified = tempMinutesToLastModified
				curTime = tempCurTime
			}
		}

		// if object >= 1000, get logs within the last 6.5 hours. Then get the new minutesToLastModified if new object > 0
		if len(objects.Contents) >= 1000 {
			now := time.Now().Unix()
			timeStamp := strconv.FormatInt(now, 10)
			pre := fmt.Sprintf("%s%s/%s", prefix, s, timeStamp[0:6])
			newObjects, err := v2S3.NewFromConfig(cfg).ListObjects(context.TODO(), &v2S3.ListObjectsInput{
				Bucket: v2Aws.String(cluster.StorageInfo[0].BucketName),
				Prefix: v2Aws.String(pre),
			})
			if err != nil {
				return lastModifiedTime, minutesToLastModified, curTime, err
			}
			if len(newObjects.Contents) > 0 {
				o := newObjects.Contents[len(newObjects.Contents)-1]
				tempMinutesToLastModified := int(tempCurTime.Sub(*o.LastModified).Minutes())
				if tempMinutesToLastModified < minutesToLastModified {
					lastModifiedTime = *o.LastModified
					minutesToLastModified = tempMinutesToLastModified
					curTime = tempCurTime
				}
			}
		}
	}
	return

}

func GetMinutesSinceLastUpdateTimeGCS(cluster *ObservabilityCluster, createTime time.Time) (lastModifiedTime time.Time, minutesToLastModified int, curTime time.Time, lastBlock string, err error) {
	prefix := fmt.Sprintf("%d/%s/%s/%s/logs/", cluster.ClusterBaseInfo.DataPropertyRights, cluster.TenantID, cluster.ProjectID, cluster.ApplicationID)

	ctx := context.Background()
	lastModifiedTime = createTime
	minutesToLastModified = int(time.Since(lastModifiedTime).Minutes())

	client, err := storage.NewClient(context.TODO())
	if err != nil {
		return
	}
	defer client.Close()

	_, cancel := context.WithTimeout(ctx, time.Second*10)
	defer cancel()

	bucket := client.Bucket(cluster.StorageInfo[0].BucketName)
	objects := bucket.Objects(context.TODO(), &storage.Query{
		Prefix: prefix,
	})

	for {
		attrs, err := objects.Next()

		if err == iterator.Done {
			break
		}
		if err != nil {
			return lastModifiedTime, minutesToLastModified, curTime, lastBlock, err
		}
		tempCurTime := time.Now()
		tempMinutesToLastModified := int(tempCurTime.Sub(attrs.Updated).Minutes())
		if tempMinutesToLastModified < minutesToLastModified {
			lastModifiedTime = attrs.Updated
			minutesToLastModified = tempMinutesToLastModified
			curTime = tempCurTime
			lastBlock = attrs.Name
		}
	}

	return
}

func GetMinutesSinceLastS3Snapshot(cluster *ObservabilityCluster, cfg v2Aws.Config, prefix string, createAt time.Time) (lastModifiedTime time.Time,
	minutesToLastModified int, err error) {

	lastBlock, _ := GetLastS3Block(cluster, cfg, prefix, []string{}, v2Aws.String(""))

	curTime := time.Now()
	//minutesSinceCreated = int(curTime.Sub(createdAt).Minutes())

	if lastBlock == "" {
		if !createAt.IsZero() {
			lastModifiedTime = createAt
			curTime = time.Now()
			minutesToLastModified = int(curTime.Sub(lastModifiedTime).Minutes())
			return
		}
	}
	lastModifiedTime = createAt
	minutesToLastModified = int(curTime.Sub(time.Time{}).Minutes())
	objects, err := v2S3.NewFromConfig(cfg).ListObjects(context.TODO(), &v2S3.ListObjectsInput{
		Bucket: v2Aws.String(cluster.StorageInfo[0].BucketName),
		Prefix: v2Aws.String(fmt.Sprintf("%s%s/", prefix, lastBlock)),
	})
	if err != nil {
		return
	}

	for _, o := range objects.Contents {

		tempMinutesToLastModified := int(curTime.Sub(*o.LastModified).Minutes())
		if tempMinutesToLastModified < minutesToLastModified {
			lastModifiedTime = *o.LastModified
			minutesToLastModified = tempMinutesToLastModified
		}
	}
	return
}

func GetMinutesSinceLastGcsSnapshot(cluster *ObservabilityCluster, createAt time.Time) (lastModifiedTime time.Time,
	minutesToLastModified int, err error) {
	prefix := fmt.Sprintf("%d/%s/%s/%s/metrics/", cluster.ClusterBaseInfo.DataPropertyRights, cluster.TenantID, cluster.ProjectID, cluster.ApplicationID)

	ctx := context.Background()
	lastModifiedTime = createAt
	minutesToLastModified = int(time.Since(lastModifiedTime).Minutes())

	client, err := storage.NewClient(context.TODO())
	if err != nil {
		return
	}
	defer client.Close()

	_, cancel := context.WithTimeout(ctx, time.Second*10)
	defer cancel()

	bucket := client.Bucket(cluster.StorageInfo[0].BucketName)
	objects := bucket.Objects(context.TODO(), &storage.Query{
		Prefix: prefix,
	})

	for {
		attrs, err := objects.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return lastModifiedTime, minutesToLastModified, err
		}
		tempCurTime := time.Now()
		tempMinutesToLastModified := int(tempCurTime.Sub(attrs.Updated).Minutes())
		if tempMinutesToLastModified < minutesToLastModified {
			lastModifiedTime = attrs.Updated
			minutesToLastModified = tempMinutesToLastModified
			//lastBlock = attrs.Name
		}
	}

	if minutesToLastModified == int(time.Since(time.Time{}).Minutes()) {
		if !createAt.IsZero() {
			lastModifiedTime = createAt
			minutesToLastModified = int(time.Since(lastModifiedTime).Minutes())
			return
		}
	}

	return
}

func GetLastS3Block(cluster *ObservabilityCluster, cfg v2Aws.Config, prefix string, lastSubPrefixes []string, marker *string) (lastBlock, msg string) {
	objects, err := v2S3.NewFromConfig(cfg).ListObjects(context.TODO(), &v2S3.ListObjectsInput{
		Bucket:    v2Aws.String(cluster.StorageInfo[0].BucketName),
		Prefix:    v2Aws.String(prefix),
		Delimiter: v2Aws.String("/"),
		Marker:    marker,
	})
	if err != nil {
		//fmt.Sprintf("[warning no bucket]%s,%s:%v", cluster.StorageInfo[0].BucketName, prefix, err)
		return "", "warning no bucket"
	}

	subPrefixes := make([]string, 0)
	for _, commonPrefix := range objects.CommonPrefixes {
		subPrefixFullPathArray := strings.Split(*commonPrefix.Prefix, "/")
		subPrefixes = append(subPrefixes, subPrefixFullPathArray[len(subPrefixFullPathArray)-2])
	}
	if len(subPrefixes) == 0 {
		if len(lastSubPrefixes) == 0 {
			return
		} else {
			return lastSubPrefixes[len(lastSubPrefixes)-1], ""
		}
	}

	sort.Strings(subPrefixes)
	if len(subPrefixes) == 1000 || *objects.IsTruncated {
		return GetLastS3Block(cluster, cfg, prefix, subPrefixes, objects.NextMarker)
	} else {
		return subPrefixes[len(subPrefixes)-1], ""
	}
}

func RequestPrometheus(url string) (*QueryResponse, error) {
	req, _ := http.NewRequest("GET", url, nil)
	resp, err := http.DefaultClient.Do(req)

	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		msg, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("promquery: expected status 2xx, got %d: %s", resp.StatusCode, string(msg))
	}

	// unmarshal resp
	respContent, _ := io.ReadAll(resp.Body)

	var res = &QueryResponse{}
	err = json.Unmarshal(respContent, res)

	return res, err
}

type QueryTiDBClusterResp struct {
	Total int64         `json:"total"`
	Items []TiDBCluster `json:"items"`
}

type TiDBCluster struct {
	TenantID         string `json:"tenantID"`
	TenantName       string `json:"tenantName"`
	TenantPlan       string `json:"tenantPlan"`
	ClusterID        string `json:"clusterID"`
	ClusterName      string `json:"clusterName"`
	ClusterCreatedAt int64  `json:"clusterCreatedAt"`
	ClusterDuration  string `json:"clusterDuration"`
	ClusterDeletedAt int64  `json:"clusterDeletedAt"`

	ClusterStatus       string `json:"clusterStatus"`
	ClusterDeployType   string `json:"clusterDeployType"`
	ClusterRegionName   string `json:"clusterRegionName"`
	ClusterProviderName string `json:"clusterProviderName"`
	ClusterShootName    string `json:"clusterShootName"`
	ClusterK8sNamespace string `json:"ClusterK8sNamespace"`
	ClusterNetworkName  string `json:"ClusterNetworkName"`
	ClusterType         string `json:"clusterType"`
}

func WithHeaderV2(request *http.Request, token string) {
	request.Header.Set("Content-Type", "application/json")
	request.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token))
}

func GenerateS3Prefix(cluster *ObservabilityCluster) string {
	c := GetConfig()
	if Contains(c.Config.OldPath, fmt.Sprintf("%s/%s/%s", cluster.TenantID, cluster.ProjectID, cluster.ApplicationID)) {
		return fmt.Sprintf("%s/tenant=%s/project=%s/application=%s/metrics/", cluster.StorageInfo[0].BucketName, cluster.TenantID, cluster.ProjectID, cluster.ApplicationID)
	} else {
		return fmt.Sprintf("%s/%d/%s/%s/%s/metrics/", cluster.StorageInfo[0].BucketName,
			cluster.ClusterBaseInfo.DataPropertyRights, cluster.TenantID, cluster.ProjectID, cluster.ApplicationID)
	}
}

func Contains(elems []string, v string) bool {
	for _, s := range elems {
		if v == s {
			return true
		}
	}
	return false
}

func clinicClusters2URL(host, clinicURL, applicationID string) string {
	q := url.Values{}
	q.Set("cluster_id", applicationID)
	q.Set("limit", "1")
	q.Set("page", "1")
	return fmt.Sprintf("https://%s%s?%s", host, clinicURL, q.Encode())
}

func GetTiDBCluster(cluster ObservabilityCluster) (*TiDBCluster, error) {
	var api string
	var clinicToken string
	if strings.HasPrefix(cluster.BizType, "staging") {
		api = clinicClusters2URL("staging-", GetConfig().Config.ClinicServer.ClinicURL, cluster.ApplicationID)
		clinicToken = GetConfig().Config.ClinicServer.ClinicTokenStaging
	} else {
		api = clinicClusters2URL("", GetConfig().Config.ClinicServer.ClinicURL, cluster.ApplicationID)
		clinicToken = GetConfig().Config.ClinicServer.ClinicToken
	}
	if GetConfig().Env == "staging" {
		api = clinicClusters2URL("staging-", GetConfig().Config.ClinicServer.ClinicURL, cluster.ApplicationID)
		clinicToken = GetConfig().Config.ClinicServer.ClinicTokenStaging
	}
	// request of querying tidb clusters
	request, _ := http.NewRequest("GET", api, nil)
	WithHeaderV2(request, clinicToken)

	// get response
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		msg, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("promquery: expected status 2xx, got %d: %s", resp.StatusCode, string(msg))
	}

	clustersResp := &QueryTiDBClusterResp{}
	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}
	err = json.Unmarshal(respContent, clustersResp)
	if err != nil {
		return nil, err
	}
	if len(clustersResp.Items) == 0 {
		return nil, nil
	}
	return &clustersResp.Items[0], nil
}

func GetTidbClusterStatus(tidbCluster *TiDBCluster, tenantID, applicationID string) (clinicStatus string, createdAt time.Time) {
	if tidbCluster == nil {
		return
	}
	if tidbCluster.TenantID == tenantID && tidbCluster.ClusterID == applicationID {
		return tidbCluster.ClusterStatus, time.Unix(0, tidbCluster.ClusterCreatedAt*int64(time.Millisecond))
	}
	return
}

func ClearFtNamespaces() {
	names, err := kubeClient.CoreV1().Namespaces().List(context.Background(), metav1.ListOptions{})
	if err != nil {
		panic(err)
	}
	for _, one := range names.Items {
		if strings.Contains(one.Name, "ft-biz-tenantid") {
			err := kubeClient.CoreV1().Namespaces().Delete(context.Background(), one.Name, metav1.DeleteOptions{})
			if err != nil {
				panic(err)
			}
		}
	}
}

func waitTimeout(wg *sync.WaitGroup, timeout time.Duration) bool {
	c := make(chan struct{})
	go func() {
		defer close(c)
		wg.Wait()
	}()
	select {
	case <-c:
		return false
	case <-time.After(timeout):
		return true
	}
}

type LokiLog struct {
	Streams []Streams `json:"streams"`
}

type Streams struct {
	Stream map[string]string `json:"stream"`
	Values [][]string        `json:"values"`
}

func (m *LokiLog) Reset()         { *m = LokiLog{} }
func (m *LokiLog) String() string { return proto.CompactTextString(m) }
func (*LokiLog) ProtoMessage()    {}
func WriteLoki(req LokiLog) (*http.Response, error) {
	pbBytes, err := proto.Marshal(&req)
	if err != nil {
		return nil, fmt.Errorf("lokiwrite: marshaling remote write request proto: %w", err)
	}

	compressedBytes := snappy.Encode(nil, pbBytes)

	// Prepare http request.
	url := "http://loki.observability.svc.cluster.local:3100/loki/api/v1/push"
	httpResp, err := http.Post(url, "application/x-protobuf", bytes.NewBuffer(compressedBytes))
	return httpResp, err
}

func QueryVM(url string) (*QueryResponse, error) {
	request, _ := http.NewRequest("GET", url, nil)
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode/100 != 2 {
		msg, _ := io.ReadAll(resp.Body)
		return nil, &WriteError{
			err:  fmt.Errorf("vmquery: expected status %d, got %d: %s", http.StatusOK, resp.StatusCode, string(msg)),
			code: resp.StatusCode,
		}
	}
	// unmarshal resp
	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}
	var res = &QueryResponse{}
	err = json.Unmarshal(respContent, res)

	return res, err
}