//go:build e2e
// +build e2e

package utils

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"math/rand"
	"net/http"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/pingcap/log"
	"gopkg.in/yaml.v2"
	"gorm.io/gorm"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	"github.com/ajg/form"
	prometheusv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
	"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/pkg/common/utils"
	mysqlclient "github.com/tidbcloud/observability/pkg/mysql/client"
	"github.com/tidbcloud/pkgv2/gormutil"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/client-go/kubernetes"
	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
	ctrl "sigs.k8s.io/controller-runtime"
)

type ConfigInfo struct {
	Service   ServiceInfo        `yaml:"service"`
	Auth0     Auth0Req           `yaml:"auth0"`
	AwsPVE    *AwsEndPoint       `yaml:"awsPVE"`
	DataLink  DataLink           `yaml:"dataLink"`
	Stress    Stress             `yaml:"stress,omitempty"`
	RDSConfig *structs.RdsConfig `yaml:"rdsConfig"`
	DSAuth    *DSAuthConfig      `yaml:"dsAuth"`
}

type DSAuthConfig struct {
	URL         string `yaml:"url"`
	GithubToken string `yaml:"githubToken"`
}

type ServiceInfo struct {
	ServerURL   string   `yaml:"serverURL"`
	RegionalURL string   `yaml:"regionalURL"`
	DsURL       string   `yaml:"dsURL"`
	Vendor      string   `yaml:"vendor"`
	Region      string   `yaml:"region"`
	UserARN     string   `yaml:"userARN"`
	Specs       []string `yaml:"specs"`
	SkipPrepare bool     `yaml:"skipPrepare"`
}

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 DSAuthReq struct {
	GithubToken string `json:"github_token"`
}

type DSAuthResp struct {
	Token string `json:"token"`
}

var config *ConfigInfo
var token string
var randomID string
var dsToken string
var kubeClient *kubernetes.Clientset
var rds *gorm.DB

func GetService() *ServiceInfo {
	if config == nil {
		panic("config is nil")
	}
	return &config.Service
}

func GetToken() *string {
	return &token
}

func GetConfig() *ConfigInfo {
	return config
}

func HandleResp(t *testing.T, resp *http.Response, data interface{}) {
	respContent, err := io.ReadAll(resp.Body)
	assert.NoError(t, err)
	err = UnmarshalRespData(respContent, data)
	assert.NoError(t, err)
}

func SetRandomID(s string) {
	randomID = s
}

func PreparConfig(path string) {
	in, err := os.ReadFile(path)
	if err != nil {
		panic(err)
	}
	var c ConfigInfo
	err = yaml.Unmarshal(in, &c)
	if err != nil {
		panic(err)
	}
	config = &c
	log.Info("find config", zap.Any("config", config))
}

func DefaultAuth0Req() Auth0Req {
	return 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 PrepareDSToken() {
	dsAuth := config.DSAuth
	if dsAuth == nil {
		log.Info("DSAuth config not set, skipping DS token preparation")
		return
	}
	if dsAuth.GithubToken == "" {
		log.Info("DSAuth githubToken not set, skipping DS token preparation")
		return
	}

	authURL := strings.TrimRight(dsAuth.URL, "/") + "/client/auth"
	reqBody := DSAuthReq{GithubToken: dsAuth.GithubToken}
	body, err := json.Marshal(reqBody)
	if err != nil {
		panic(fmt.Sprintf("failed to marshal DS auth request: %v", err))
	}

	resp, err := http.Post(authURL, "application/json", strings.NewReader(string(body)))
	if err != nil {
		panic(fmt.Sprintf("failed to call DS auth endpoint: %v", err))
	}
	defer resp.Body.Close()

	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		panic(fmt.Sprintf("failed to read DS auth response: %v", err))
	}

	if resp.StatusCode != http.StatusOK {
		panic(fmt.Sprintf("DS auth failed with status %d: %s", resp.StatusCode, string(respContent)))
	}

	dsResp := &DSAuthResp{}
	err = json.Unmarshal(respContent, dsResp)
	if err != nil {
		panic(fmt.Sprintf("failed to unmarshal DS auth response: %v", err))
	}
	dsToken = dsResp.Token
	log.Info("DS token prepared successfully")
}

func GetDSToken() string {
	return dsToken
}

func WithDSAuthHeader(request *http.Request) {
	request.Header.Set("Content-Type", "application/json")
	request.Header.Set("Accept-Encoding", "identity")
	if dsToken != "" {
		request.Header.Set(constants.AuthorizationHeader, fmt.Sprintf("%s %s", constants.BearerPrefix, dsToken))
	}
	s := GetService()
	request.Header.Set(string(constants.OpenAPIHeaderVendor), s.Vendor)
	request.Header.Set(string(constants.OpenAPIHeaderRegion), s.Region)
}

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

func PrepareRDS() {
	cfg := config.RDSConfig.ToGormutilConfig()
	db, err := gormutil.Connect(*cfg)
	if err != nil {
		log.Fatal("failed to connect to MySQL", zap.Error(err))
	}
	sqlDB, err := db.DB()
	if err != nil {
		log.Fatal("failed to get sql.DB from gorm.DB", zap.Error(err))
	}
	if err := sqlDB.Ping(); err != nil {
		log.Fatal("failed to ping database", zap.Error(err))
	}

	rds = db
}

const (
	DefaultBizType     = "ft-biz"
	PerfBizType        = "dev-devtier"
	ServerlessBizType  = "dev-devtier"
	DedicatedBizType   = "tidbcloud"
	NexgenBizType      = "dev-tidbcloud-nextgen"
	RuleTmplServerless = "tidb-cloud-control-plane-dev-tier-rule"
)

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, DefaultBizType) {
			fmt.Printf("delete namespace: %s\n", one.Name)
			err := kubeClient.CoreV1().Namespaces().Delete(context.Background(), one.Name, metav1.DeleteOptions{})
			if err != nil {
				panic(err)
			}
		}
	}
}

func ClearNamespacesForPerformanceTest() {
	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, "dev-devtier-perf-tenant") {
			fmt.Printf("delete namespace: %s\n", one.Name)
			err := kubeClient.CoreV1().Namespaces().Delete(context.Background(), one.Name, metav1.DeleteOptions{})
			if err != nil {
				panic(err)
			}
		}
	}
}

func DeletePod(nameSpace, labelSelector string) {
	err := kubeClient.CoreV1().Pods(nameSpace).DeleteCollection(
		context.TODO(),
		metav1.DeleteOptions{},
		metav1.ListOptions{
			LabelSelector: labelSelector,
		},
	)
	if err != nil {
		log.Fatal("Failed to delete pod", zap.Error(err))
	}
}

var ApiPrefix = "/api/v1"

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

func DefaultCreateReq() message.CreateClusterReq {
	s := GetService()
	return message.CreateClusterReq{
		ClusterBaseInfo: structs.ClusterBaseInfo{
			CommonReqInfo: structs.CommonReqInfo{
				Vendor: constants.ProvidedVendor(s.Vendor),
				Region: s.Region,
			},
			TenantID:           randomID,
			ProjectID:          randomID,
			ApplicationID:      randomID,
			BizType:            DefaultBizType,
			Version:            "v1",
			ShareMode:          constants.EXCLUSIVE,
			Tags:               []string{"tag1", "default"},
			DataPropertyRights: constants.DataPropertyRights(rand.Intn(2)),
		},
		AgentAccessARN: s.UserARN,
		MonitorSpecID:  s.Specs[rand.Intn(len(s.Specs))],
	}
}

func DefaultCreateSharedReq(tenant string, project string, application string) message.CreateClusterReq {
	s := GetService()
	return message.CreateClusterReq{
		ClusterBaseInfo: structs.ClusterBaseInfo{
			CommonReqInfo: structs.CommonReqInfo{
				Vendor: constants.ProvidedVendor(s.Vendor),
				Region: s.Region,
			},
			TenantID:      tenant,
			ProjectID:     project,
			ApplicationID: application,
			BizType:       DefaultBizType,
			Version:       "v1",
			ShareMode:     constants.SHARED,
			Tags:          []string{"tag1", "default"},
		},
		AgentAccessARN: s.UserARN,
		MonitorSpecID:  "vmalert.default",
	}
}

func DefaultCreateSharedReqForPerformanceTest(tenant string, project string, application string) message.CreateClusterReq {
	s := GetService()
	return message.CreateClusterReq{
		ClusterBaseInfo: structs.ClusterBaseInfo{
			CommonReqInfo: structs.CommonReqInfo{
				Vendor: constants.ProvidedVendor(s.Vendor),
				Region: s.Region,
			},
			TenantID:      tenant,
			ProjectID:     project,
			ApplicationID: application,
			BizType:       PerfBizType,
			Version:       "v1",
			ShareMode:     constants.SHARED,
			Tags:          []string{"tag1", "default"},
		},
		AgentAccessARN:  s.UserARN,
		MonitorSpecID:   s.Specs[rand.Intn(len(s.Specs))],
		KeySpaceID:      fmt.Sprintf("%d", rand.Int31()),
		ParentClusterID: "76fcbc66-1b24-478e-a13c-04a249f2b239",
	}
}

func DefaultCreateServerlessParentReq(tenant string, project string, application string) message.CreateClusterReq {
	s := GetService()
	return message.CreateClusterReq{
		ClusterBaseInfo: structs.ClusterBaseInfo{
			CommonReqInfo: structs.CommonReqInfo{
				Vendor: constants.ProvidedVendor(s.Vendor),
				Region: s.Region,
			},
			TenantID:      tenant,
			ProjectID:     project,
			ApplicationID: application,
			BizType:       ServerlessBizType,
			Version:       "v1",
			ShareMode:     constants.SHARED,
			Tags:          []string{"tag1", "default"},
		},
		AgentAccessARN: s.UserARN,
		MonitorSpecID:  "vmalert.default",
		RuleTemplateID: RuleTmplServerless,
	}
}

func DefaultCreateServerlessChildReq(tenant string, project string, application string, parentClusterID string, keyspaceID string) message.CreateClusterReq {
	s := GetService()
	return message.CreateClusterReq{
		ClusterBaseInfo: structs.ClusterBaseInfo{
			CommonReqInfo: structs.CommonReqInfo{
				Vendor: constants.ProvidedVendor(s.Vendor),
				Region: s.Region,
			},
			TenantID:      tenant,
			ProjectID:     project,
			ApplicationID: application,
			BizType:       ServerlessBizType,
			Version:       "v1",
			ShareMode:     constants.SHARED,
		},
		AgentAccessARN:  s.UserARN,
		MuteAlert:       true,
		ParentClusterID: parentClusterID,
		KeySpaceID:      keyspaceID,
	}
}

func HandleWithDefault(handle func(req message.CreateClusterReq) message.CreateClusterReq) message.CreateClusterReq {
	return handle(DefaultCreateSharedReq("111", "222", "333"))
}

func HandleWithServerlessParent(handle func(req message.CreateClusterReq) message.CreateClusterReq) message.CreateClusterReq {
	return handle(DefaultCreateServerlessParentReq("devtier-ft", "ft", "devtier-ft"))
}

func QueryReq(req message.QueryClustersReq) *http.Request {
	s := GetService()
	queryString := fmt.Sprintf("%s%s/clusters/?vendor=%s&region=%s&tenantID=%s&projectID=%s&applicationID=%s&limit=%d&continueToken=%s",
		s.ServerURL, ApiPrefix,
		s.Vendor, s.Region,
		req.TenantID, req.ProjectID, req.ApplicationID,
		req.Limit, req.ContinueToken)
	if req.BizType != "" {
		queryString = fmt.Sprintf("%s&bizType=%s", queryString, req.BizType)
	}
	request, _ := http.NewRequest("GET", queryString, nil)
	WithHeader(request)

	return request
}

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

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

func DeleteReq(req message.DeleteClusterReq) *http.Request {
	api := fmt.Sprintf("%s%s/clusters/%s?vendor=%s&region=%s",
		GetService().ServerURL, ApiPrefix, req.ID, req.Vendor, req.Region)

	body, _ := json.Marshal(req)
	request, _ := http.NewRequest("DELETE", api, strings.NewReader(string(body)))
	if req.BizType != "" {
		request.Header.Set("X-Biz-Type", req.BizType)
	}
	WithHeader(request)
	return request
}

func DetailReq(req message.DetailClusterReq) *http.Request {
	api := fmt.Sprintf("%s%s/clusters/%s?vendor=%s&region=%s",
		GetService().ServerURL, ApiPrefix, req.ID, req.Vendor, req.Region)

	body, _ := json.Marshal(req)
	request, _ := http.NewRequest("GET", api, strings.NewReader(string(body)))
	// Add bizType if provided
	if req.BizType != "" {
		request.Header.Set("X-Biz-Type", req.BizType)
	}
	WithHeader(request)
	return request
}

func ModifySpecReq(req message.ModifyMonitorSpecReq) *http.Request {
	api := fmt.Sprintf("%s%s/clusters/%s/monitor-spec/?vendor=%s&region=%s",
		GetService().ServerURL, ApiPrefix, req.ID, req.Vendor, req.Region)

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

func PauseClusterReq(req message.PauseClusterReq) *http.Request {
	api := fmt.Sprintf("%s%s/clusters/%s/pause/?id=%s&vendor=%s&region=%s",
		GetService().ServerURL, ApiPrefix, req.ID, req.ID, req.Vendor, req.Region)

	body, _ := json.Marshal(req)
	request, _ := http.NewRequest("PUT", api, strings.NewReader(string(body)))
	WithHeader(request)
	fmt.Printf("request pause prom, uri is %s, header is %s, body is %s, request method is %s", request.RequestURI, request.Header, request.Body, request.Method)
	return request
}

func ResumeClusterReq(req message.ResumeClusterReq) *http.Request {
	api := fmt.Sprintf("%s%s/clusters/%s/resume?id=%s&vendor=%s&region=%s",
		GetService().ServerURL, ApiPrefix, req.ID, req.ID, req.Vendor, req.Region)

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

func QueryClusterReq(req message.DetailClusterReq) *http.Request {
	api := fmt.Sprintf("%s%s/clusters/%s?id=%s&vendor=%s&region=%s",
		GetService().ServerURL, ApiPrefix, req.ID, req.ID, req.Vendor, req.Region)

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

func WithHeader(request *http.Request) {
	request.Header.Set("Content-Type", "application/json")
	request.Header.Set("Accept-Encoding", "identity") // disable gzip
	request.Header.Set(constants.AuthorizationHeader, fmt.Sprintf("%s %s", constants.BearerPrefix, token))

}

func CheckStatus(t *testing.T, id string, getFn func(string) *structs.ObservabilityCluster, intermediateStatus, finalStatus constants.ResourceStatus) {
	assert.Contains(t, []constants.ResourceStatus{intermediateStatus, finalStatus, ""}, getFn(id).Status)

	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()

	count := 60

	for range ticker.C {
		cluster := getFn(id)
		assert.Contains(t, []constants.ResourceStatus{intermediateStatus, finalStatus, ""}, cluster.Status)

		if cluster.Status == finalStatus {
			return
		}
		count--
		if count == 0 {
			assert.Equal(t, "succeed", "timeout")
		}
	}
}

func CreateCluster(req message.CreateClusterReq) *message.CreateClusterResp {
	request := CreateReq(req)
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()
	clustersResp := &message.CreateClusterResp{}
	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		panic(err)
	}
	if resp.StatusCode != 202 && resp.StatusCode != 409 {
		panic(string(respContent))
	}
	err = UnmarshalRespData(respContent, clustersResp)
	if err != nil {
		panic(err)
	}
	return clustersResp
}

func DeleteCluster(id string) *http.Response {
	return DeleteClusterWithBizType(id, "")
}

func DeleteServerlessCluster(id string) *http.Response {
	return DeleteClusterWithBizType(id, ServerlessBizType)
}

func DeleteClusterWithBizType(id string, bizType string) *http.Response {
	s := GetService()
	request := DeleteReq(message.DeleteClusterReq{
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(s.Vendor),
			Region: s.Region,
		},
		ID:      id,
		BizType: bizType,
	})
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		panic(err)
	}
	return resp
}

func SoftDeleteCluster(id string) *http.Response {
	s := GetService()
	request := DeleteReq(message.DeleteClusterReq{
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(s.Vendor),
			Region: s.Region,
		},
		ID: id,
	})
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		panic(err)
	}
	return resp
}

func GetClusterIfExisted(id string) (bool, *structs.ObservabilityCluster) {
	return GetClusterIfExistedWithBizType(id, "")
}

func GetServerlessClusterIfExisted(id string) (bool, *structs.ObservabilityCluster) {
	return GetClusterIfExistedWithBizType(id, ServerlessBizType)
}

func GetClusterIfExistedWithBizType(id string, bizType string) (bool, *structs.ObservabilityCluster) {
	s := GetService()

	request := DetailReq(message.DetailClusterReq{
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(s.Vendor),
			Region: s.Region,
		},
		ID:      id,
		BizType: bizType,
	})
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()

	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		panic(err)
	}

	if resp.StatusCode == http.StatusNotFound {
		fmt.Printf("Cluster %s not found (404). Response: %s\n", id, string(respContent))
		return false, nil
	}

	if resp.StatusCode != http.StatusOK {
		fmt.Printf("Unexpected status %d for cluster %s. Response: %s\n", resp.StatusCode, id, string(respContent))
		return false, nil
	}

	clustersResp := &message.DetailClusterResp{}
	err = UnmarshalRespData(respContent, clustersResp)
	if err != nil {
		fmt.Printf("Failed to unmarshal response for cluster %s: %v. Response: %s\n", id, err, string(respContent))
		panic(err)
	}
	return true, &clustersResp.ObservabilityCluster
}

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 GetCluster(id string) *structs.ObservabilityCluster {
	return GetClusterWithBizType(id, "")
}

func GetServerlessCluster(id string) *structs.ObservabilityCluster {
	return GetClusterWithBizType(id, ServerlessBizType)
}

func GetClusterWithBizType(id string, bizType string) *structs.ObservabilityCluster {
	existed, cluster := GetClusterIfExistedWithBizType(id, bizType)
	if existed {
		return cluster
	} else {
		panic(fmt.Sprintf("cluster id not found, %s", id))
	}
}

func ModifySpec(id, spec string) *http.Response {
	s := GetService()
	request := ModifySpecReq(message.ModifyMonitorSpecReq{
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(s.Vendor),
			Region: s.Region,
		},
		ID:     id,
		SpecID: spec,
	})
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		panic(err)
	}
	return resp
}

func PauseCluster(id string) *http.Response {
	s := GetService()
	request := PauseClusterReq(message.PauseClusterReq{
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(s.Vendor),
			Region: s.Region,
		},
		ID: id,
	})

	var (
		resp      *http.Response
		err       error
		errCnt    = 0
		maxErrCnt = 3
	)

	for {
		resp, err = http.DefaultClient.Do(request)
		fmt.Printf("pause cluster,response is %v\n", resp)
		if err != nil {
			errCnt++
			if errCnt < maxErrCnt {
				time.Sleep(3 * time.Second)
				continue
			}
			panic(err)
		}

		if resp.StatusCode == http.StatusAccepted {
			break
		}

		time.Sleep(3 * time.Second)
		continue
	}

	return resp
}

func ResumeCluster(id string) *http.Response {
	s := GetService()
	request := ResumeClusterReq(message.ResumeClusterReq{
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(s.Vendor),
			Region: s.Region,
		},
		ID: id,
	})
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		panic(err)
	}
	return resp
}

func QueryCluster(id string) *message.DetailClusterResp {
	s := GetService()
	request := QueryClusterReq(message.DetailClusterReq{
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(s.Vendor),
			Region: s.Region,
		},
		ID: id,
	})
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()

	result := &message.DetailClusterResp{}

	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		panic(err)
	}

	err = UnmarshalRespData(respContent, result)
	if err != nil {
		panic(err)
	}
	return result
}

func QueryPrometheusRuleReq(req message.QueryAlertRulesReq) *http.Request {
	api := fmt.Sprintf("%s%s/external/%s/orgs/%s/projects/%s/clusters/%s/monitor-alertrules?vendor=%s&region=%s",
		GetService().RegionalURL, ApiPrefix, req.BizType, req.TenantID, req.ProjectID, req.ApplicationID, req.Vendor, req.Region)
	log.Info("QueryPrometheusRuleReq", zap.String("api", api))
	request, _ := http.NewRequest("GET", api, nil)
	WithHeader(request)
	return request
}

func DeleteTemplatePrometheusRuleReq(req message.DeleteAlertRulesTemplateReq) *http.Request {
	api := fmt.Sprintf("%s%s/alertrule-template/%s/?vendor=%s&region=%s",
		GetService().ServerURL, ApiPrefix, req.ID, req.Vendor, req.Region)

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

func UpdateClusterRuleReq(req message.UpdateAlertRulesByCloudUserReq) *http.Request {
	api := fmt.Sprintf("%s%s/external/%s/orgs/%s/projects/%s/clusters/%s/monitor-alertrules?vendor=%s&region=%s",
		GetService().RegionalURL, ApiPrefix, req.BizType, req.TenantID, req.ProjectID, req.ApplicationID, req.Vendor, req.Region)

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

func QueryClusterAlertsReq(req message.QueryExternalAlertsReq) *http.Request {
	api := fmt.Sprintf("%s%s/external/%s/orgs/%s/projects/%s/clusters/%s/alerts?vendor=%s&region=%s&limit=20&page=1&status=%s&startTime=%d&endTime=%d",
		GetService().RegionalURL, ApiPrefix, req.BizType, req.TenantID, req.ProjectID, req.ApplicationID, req.Vendor, req.Region, string(req.State), *req.StartTime, *req.EndTime)

	request, _ := http.NewRequest("GET", api, nil)
	WithHeader(request)
	return request
}

func ModifyPrometheusRuleTemplateReq(req message.ModifyAlertRulesTemplateReq) *http.Request {
	api := fmt.Sprintf("%s%s/alertrule-template/",
		GetService().ServerURL, ApiPrefix)
	body, _ := json.Marshal(req)
	request, _ := http.NewRequest("POST", api, strings.NewReader(string(body)))
	WithHeader(request)
	return request
}
func QueryPrometheusTemplateReq(req message.QueryAlertRulesTemplateReq) *http.Request {
	api := fmt.Sprintf("%s%s/alertrule-template/%s/?vendor=%s&region=%s",
		GetService().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 DeletePrometheusTemplate(id string) *message.DeleteAlertRulesTemplateResp {
	s := GetService()
	request := DeleteTemplatePrometheusRuleReq(message.DeleteAlertRulesTemplateReq{
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(s.Vendor),
			Region: s.Region,
		},
		ID: id,
	})
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()

	result := &message.DeleteAlertRulesTemplateResp{}

	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		panic(err)
	}

	err = UnmarshalRespData(respContent, result)
	if err != nil {
		panic(err)
	}
	return result
}

func QueryPrometheusTemplate(id string) *message.QueryAlertRulesTemplateResp {
	s := GetService()
	request := QueryPrometheusTemplateReq(message.QueryAlertRulesTemplateReq{
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(s.Vendor),
			Region: s.Region,
		},
		ID: id,
	})
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()

	result := &message.QueryAlertRulesTemplateResp{}

	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		panic(err)
	}

	err = UnmarshalRespData(respContent, result)
	if err != nil {
		panic(err)
	}
	return result
}

func ModifyPrometheusRuleTemplate(id string, op string, opts ...RuleOption) *http.Response {
	s := GetService()

	template := DefaultRuleTemplate()
	for _, opt := range opts {
		opt(&template)
	}

	request := ModifyPrometheusRuleTemplateReq(message.ModifyAlertRulesTemplateReq{
		TemplateID: id,
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(s.Vendor),
			Region: s.Region,
		},
		MonitorPromRules: template,
	})
	if op == "update" {
		request = ModifyPrometheusRuleTemplateReq(message.ModifyAlertRulesTemplateReq{
			TemplateID: id,
			CommonReqInfo: structs.CommonReqInfo{
				Vendor: constants.ProvidedVendor(s.Vendor),
				Region: s.Region,
			},
			MonitorPromRules: prometheusv1.PrometheusRuleSpec{
				Groups: []prometheusv1.RuleGroup{
					0: {
						Name: "test",
						Rules: []prometheusv1.Rule{
							{
								Alert: "HighRequestLatency",
								Expr:  intstr.FromString("request_latency_seconds:mean5m{job=\"myjob\"} > 0.5"),
								For:   "10m",
								Labels: map[string]string{
									"severity":   "critical",
									"visibility": "external",
									"ruleID":     "5",
									"status":     "active",
								},
								Annotations: map[string]string{
									"summary": "High request latency",
								},
							},
						},
					},
					1: {
						Name: "test2",
						Rules: []prometheusv1.Rule{
							{
								Alert: "HighRequestLatency2",
								Expr:  intstr.FromString("request_latency_seconds:mean5m{job=\"myjob\"} > 0.5"),
								For:   "10m",
								Labels: map[string]string{
									"severity":   "critical",
									"visibility": "external",
									"ruleID":     "5",
									"status":     "active",
								},
								Annotations: map[string]string{
									"summary": "High request latency",
								},
							},
						},
					},
				},
			},
		})
	}
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		panic(err)
	}
	return resp
}

func QueryPrometheusRule(bizType, tenantID, projectID, applicationID string) (*message.QueryAlertRulesByCloudUserResp, error) {
	s := GetService()
	request := QueryPrometheusRuleReq(message.QueryAlertRulesReq{
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(s.Vendor),
			Region: s.Region,
		},
		BizType:       bizType,
		TenantID:      tenantID,
		ProjectID:     projectID,
		ApplicationID: applicationID,
	})
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	result := &message.QueryAlertRulesByCloudUserResp{}

	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}

	err = UnmarshalRespData(respContent, result)
	if err != nil {
		log.Error("parsing response", zap.String("Response", string(respContent)))
		return nil, err
	}
	return result, nil
}

func QueryClusterAlerts(bizType, tenantID, projectID, applicationID string, status structs.AlertState) (*message.QueryExternalAlertsResp, error) {
	s := GetService()
	request := QueryClusterAlertsReq(message.QueryExternalAlertsReq{
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(s.Vendor),
			Region: s.Region,
		},
		BizType:       bizType,
		TenantID:      tenantID,
		ProjectID:     projectID,
		ApplicationID: applicationID,
		State:         status,
		StartTime:     utils.Int64Ptr(time.Now().Add(-time.Hour).Unix()),
		EndTime:       utils.Int64Ptr(time.Now().Unix()),
	})
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode/100 != 2 {
		return nil, fmt.Errorf("unexpected status code %d", resp.StatusCode)
	}
	result := &message.QueryExternalAlertsResp{}

	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}

	err = UnmarshalRespData(respContent, result)
	if err != nil {
		log.Error("parsing response", zap.String("Response", string(respContent)))
		return nil, err
	}
	return result, nil
}

func UpdateClusterPrometheusRule(bizType, tenantID, projectID, applicationID string, rules map[string]structs.AlertRuleBase) *http.Response {
	s := GetService()
	request := UpdateClusterRuleReq(message.UpdateAlertRulesByCloudUserReq{
		BizType:       bizType,
		TenantID:      tenantID,
		ProjectID:     projectID,
		ApplicationID: applicationID,
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(s.Vendor),
			Region: s.Region,
		},
		AlertRules: rules,
	})
	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		panic(err)
	}

	return resp
}

// CheckClusterInDatabase verifies that a cluster exists in the database with correct records
func CheckClusterInDatabase(t *testing.T, clusterID string) error {
	if rds == nil {
		return fmt.Errorf("RDS connection not initialized")
	}

	// Check Cluster table
	var cluster mysqlclient.Cluster
	result := rds.Where("cluster_id = ?", clusterID).First(&cluster)
	if result.Error != nil {
		return fmt.Errorf("cluster not found in database: %w", result.Error)
	}

	// Verify basic fields
	if cluster.ClusterID != clusterID {
		return fmt.Errorf("cluster_id mismatch: expected %s, got %s", clusterID, cluster.ClusterID)
	}

	// Check ClusterSpec table
	var spec mysqlclient.ClusterSpec
	result = rds.Where("cluster_id = ?", clusterID).First(&spec)
	if result.Error != nil {
		return fmt.Errorf("cluster spec not found in database: %w", result.Error)
	}

	// Check ClusterStatus table
	var status mysqlclient.ClusterStatus
	result = rds.Where("cluster_id = ?", clusterID).First(&status)
	if result.Error != nil {
		return fmt.Errorf("cluster status not found in database: %w", result.Error)
	}

	t.Logf("Database check passed for cluster %s: namespace=%s, biz_type=%s, tenant=%s, phase=%s",
		clusterID, cluster.Namespace, cluster.BizType, cluster.Tenant, status.Phase)

	return nil
}

// CheckClusterEventInDatabase verifies that a cluster event exists in the database
func CheckClusterEventInDatabase(t *testing.T, namespace, name, eventType string) error {
	if rds == nil {
		return fmt.Errorf("RDS connection not initialized")
	}

	var event mysqlclient.ClusterEvent
	result := rds.Where("namespace = ? AND name = ? AND event_type = ?", namespace, name, eventType).
		Order("created_at DESC").
		First(&event)

	if result.Error != nil {
		return fmt.Errorf("cluster event not found: namespace=%s, name=%s, type=%s, error=%w",
			namespace, name, eventType, result.Error)
	}

	t.Logf("Found cluster event: id=%d, namespace=%s, name=%s, type=%s, status=%s",
		event.EventID, event.Namespace, event.Name, event.EventType, event.Status)

	return nil
}

// CheckClusterDeletedFromDatabase verifies that a cluster has been deleted from the database
func CheckClusterDeletedFromDatabase(t *testing.T, clusterID string) error {
	if rds == nil {
		return fmt.Errorf("RDS connection not initialized")
	}

	// Check if cluster still exists
	var cluster mysqlclient.Cluster
	result := rds.Where("cluster_id = ?", clusterID).First(&cluster)

	if result.Error == nil {
		// Cluster still exists, check if deletion_timestamp is set
		if cluster.DeletionTimestamp != nil {
			t.Logf("Cluster %s has deletion_timestamp set: %v", clusterID, cluster.DeletionTimestamp)
			return nil // Soft delete is acceptable
		}
		return fmt.Errorf("cluster %s still exists in database without deletion_timestamp", clusterID)
	}

	// Check if error is "record not found" (expected)
	if result.Error.Error() == "record not found" {
		t.Logf("Cluster %s successfully deleted from database", clusterID)
		return nil
	}

	return fmt.Errorf("unexpected error checking cluster deletion: %w", result.Error)
}

// GetClusterFromDatabase retrieves a cluster record directly from the database
func GetClusterFromDatabase(clusterID string) (*mysqlclient.Cluster, error) {
	if rds == nil {
		return nil, fmt.Errorf("RDS connection not initialized")
	}

	var cluster mysqlclient.Cluster
	result := rds.Preload("Spec").Preload("Status").Where("cluster_id = ?", clusterID).First(&cluster)
	if result.Error != nil {
		return nil, result.Error
	}

	return &cluster, nil
}

// QueryClusters queries clusters with the given parameters and returns the response
func QueryClusters(req message.QueryClustersReq) (*message.QueryClustersResp, error) {
	request := QueryReq(req)

	httpResp, err := http.DefaultClient.Do(request)
	if err != nil {
		return nil, fmt.Errorf("failed to execute query request: %w", err)
	}
	defer httpResp.Body.Close()

	if httpResp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(httpResp.Body)
		return nil, fmt.Errorf("query failed with status %d: %s", httpResp.StatusCode, string(body))
	}

	respContent, err := io.ReadAll(httpResp.Body)
	if err != nil {
		return nil, fmt.Errorf("failed to read response: %w", err)
	}

	var queryResp message.QueryClustersResp
	err = UnmarshalRespData(respContent, &queryResp)
	if err != nil {
		return nil, fmt.Errorf("failed to unmarshal response: %w", err)
	}

	return &queryResp, nil
}