//go:build e2e
// +build e2e

package performance

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"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"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"k8s.io/apimachinery/pkg/util/rand"
)

func getLogger(file io.Writer) zap.Logger {
	encoder := zapcore.NewConsoleEncoder(zap.NewDevelopmentEncoderConfig())
	writer := zapcore.AddSync(file)
	core := zapcore.NewCore(encoder, writer, zapcore.DebugLevel)
	return *zap.New(core)
}

func TestWritePerformance(t *testing.T) {
	if err := os.Remove("created_clusters.txt"); err != nil && !os.IsNotExist(err) {
		t.Fatalf("Failed to remove created_clusters.txt: %v", err)
	}
	// clear
	mu.Lock()
	wrongMap = make(map[string]int)
	mu.Unlock()

	createStatusStats = &StatusCodeStats{counts: make(map[int]int)}
	createTimeStats = &DurationStats{}
	atomic.StoreUint64(&clusterCreationCounter, 0)

	// Generate random prefix for this test run
	randomPrefix := time.Now().Unix() % 1e6
	t.Logf("Starting write performance test with random prefix: %d, totalTenants=%d, clustersPerTenant=%d, bigTenantEnabled=%t",
		randomPrefix, totalTenants, clustersPerTenant, enableBigTenant)

	// Save the random prefix to file for read tests
	if err := saveRandomPrefixToFile("created_clusters.txt", fmt.Sprintf("%d", randomPrefix)); err != nil {
		t.Fatalf("Failed to save random prefix: %v", err)
	}
	f, _ := os.OpenFile(fmt.Sprintf("log/error_TestWritePerformance_%s.log", time.Now().Format("20060102_150405")), os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0666)

	log := getLogger(f)

	type workItem struct {
		tenantIdx  int
		clusterIdx int
	}

	var workItems []workItem
	tenantBigIdx := randomPrefix + 1e7
	tbClusterId := 0

	for clusterIdx := 0; clusterIdx < clustersPerTenant; clusterIdx++ {
		for tenantIdx := 0; tenantIdx < totalTenants; tenantIdx++ {
			workItems = append(workItems, workItem{tenantIdx: tenantIdx, clusterIdx: clusterIdx})
			// items of tenantBig is equal to 1/4 of the rest items
			if enableBigTenant && tbClusterId%4 == 0 {
				workItems = append(workItems, workItem{tenantIdx: int(tenantBigIdx), clusterIdx: tbClusterId})
				tbClusterId++
			}
		}
	}

	totalItems := len(workItems) // tenant * clustersPerTenant
	batchSize := 100
	if writeBatchSize > 0 {
		batchSize = writeBatchSize
	}

	// print process
	startTime := time.Now()
	progressDone := make(chan struct{})
	progressCtx, progressCancel := context.WithCancel(context.Background())

	go func() {
		defer close(progressDone)
		progressTicker := time.NewTicker(300 * time.Second) // 5 min
		defer progressTicker.Stop()

		for {
			select {
			case <-progressCtx.Done():
				return
			case <-progressTicker.C:
				completed := int(atomic.LoadUint64(&clusterCreationCounter))
				percentComplete := float64(completed) / float64(totalItems) * 100
				elapsed := time.Since(startTime)

				var remainingTime time.Duration
				if completed > 0 {
					timePerItem := elapsed / time.Duration(completed)
					remainingItems := totalItems - completed
					remainingTime = time.Duration(remainingItems) * timePerItem
				}

				t.Logf("Progress: %.1f%% | Completed: %d/%d | Elapsed: %v | Est. Remaining: %v ",
					percentComplete, completed, totalItems, elapsed.Round(time.Second),
					remainingTime.Round(time.Second))
			}
		}
	}()

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

	for i := 0; i < totalItems; i += batchSize {
		<-ticker.C
		end := i + batchSize
		if end > totalItems {
			end = totalItems
		}
		batch := workItems[i:end]

		var wg sync.WaitGroup
		wg.Add(len(batch))

		for _, work := range batch {
			go func(w workItem) {
				defer wg.Done()

				tenantID := fmt.Sprintf("pt-%d-%d", randomPrefix, w.tenantIdx)
				projectID := fmt.Sprintf("pp-%d-%d", w.tenantIdx, w.clusterIdx)
				applicationID := fmt.Sprintf("pa-%d-%d-%d", randomPrefix, w.tenantIdx, w.clusterIdx)

				var req message.CreateClusterReq
				if testBizType == utils.ServerlessBizType {
					req = utils.DefaultCreateServerlessChildReq(tenantID, projectID, applicationID, parentClusterID, fmt.Sprintf("%d", int32(time.Now().Unix())))
				} else {
					req = utils.HandleWithDefault(func(req message.CreateClusterReq) message.CreateClusterReq {
						req.TenantID = tenantID
						req.ProjectID = projectID
						req.ApplicationID = applicationID
						return req
					})
				}
				request := utils.CreateReq(req)

				requestCtx, requestCancel := context.WithTimeout(context.Background(), rwInterval)

				startReqTime := time.Now()
				resp, err := perfClient.Do(request.WithContext(requestCtx))
				defer requestCancel()
				duration := time.Since(startReqTime)

				if err == nil && resp.StatusCode == 400 {
					for i := 0; i < 3; i++ {
						if resp != nil {
							io.Copy(io.Discard, resp.Body)
							resp.Body.Close()
						}
						time.Sleep(100 * time.Millisecond)
						requestCtx, requestCancel := context.WithTimeout(context.Background(), rwInterval)
						startReqTime := time.Now()
						request := utils.CreateReq(req) // create request again to void body is empty
						resp, err = perfClient.Do(request.WithContext(requestCtx))
						duration = time.Since(startReqTime)
						requestCancel()
						if err != nil || resp.StatusCode != 400 {
							break
						}
						time.Sleep((1 * time.Second))
					}
				}

				var finalStatusCode int
				var finalDuration time.Duration
				var clusterID string
				if err != nil {
					status := -1
					if resp != nil {
						status = resp.StatusCode
					}
					key := fmt.Sprintf("%s:%d", err.Error(), status)
					mu.Lock()
					wrongMap[key]++
					mu.Unlock()
					if errors.Is(err, context.DeadlineExceeded) {
						finalStatusCode = 999 // timeout
						finalDuration = rwInterval + 100*time.Millisecond
					} else {
						finalStatusCode = -1
						finalDuration = duration
					}
					log.Error("error", zap.Error(err))
				} else {
					//fmt.Println(resp)
					finalStatusCode = resp.StatusCode
					if finalStatusCode >= 200 && finalStatusCode < 300 {
						clusterResp := &message.CreateClusterResp{}
						utils.HandleResp(t, resp, clusterResp)
						clusterID = clusterResp.ID
						createdClustersMu.Lock()
						createdClusters = append(createdClusters, clusterID)
						createdClustersMu.Unlock()

						if err := saveClusterIDToFile("created_clusters.txt", clusterID); err != nil {
							log.Error("Failed to save cluster id",
								zap.Error(err),
								zap.String("Cluster_ID", clusterID))
						}
					}
					io.Copy(io.Discard, resp.Body)
					resp.Body.Close()
					finalDuration = duration
				}

				createStatusStats.Record(finalStatusCode)
				createTimeStats.Record(finalDuration)
				atomic.AddUint64(&clusterCreationCounter, 1)
			}(work)
		}
		wg.Wait()
	}
	progressCancel()
	<-progressDone

	totalTime := time.Since(startTime)
	totalCreated := int(atomic.LoadUint64(&clusterCreationCounter))
	printPerformanceStats(t, "TestWritePerformance", "Cluster Creation", createStatusStats, createTimeStats, totalCreated, wrongMap)
	t.Logf("Total creation time: %v", totalTime)

	assert.Greater(t, len(createdClusters), 0, "Should successfully create at least some clusters")
}

func TestNormalReadPerformance(t *testing.T) {
	createdClusters, _ = loadClusterIDs("created_clusters.txt")
	if len(createdClusters) == 0 {
		t.Fatal("No clusters available for read test, please run write test first")
	}

	randomPrefix, err := loadRandomPrefix("created_clusters.txt")
	if err != nil {
		t.Fatalf("Failed to load random prefix: %v", err)
	}
	t.Logf("Loaded random prefix for read test: %s", randomPrefix)

	// clear
	mu.Lock()
	wrongMap = make(map[string]int)
	mu.Unlock()

	listStatusStats = &StatusCodeStats{counts: make(map[int]int)}
	listTimeStats = &DurationStats{}
	detailStatusStats = &StatusCodeStats{counts: make(map[int]int)}
	detailTimeStats = &DurationStats{}
	atomic.StoreUint64(&listQueryCounter, 0)
	atomic.StoreUint64(&detailQueryCounter, 0)

	t.Log("Starting normal read performance test...")
	f, _ := os.OpenFile(fmt.Sprintf("log/error_TestNormalReadPerformance_List_%s.log", time.Now().Format("20060102_150405")), os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0666)

	log := getLogger(f)

	listCtx, listCancel := context.WithTimeout(context.Background(), ctlreadTestDuration)
	defer listCancel()

	// print process
	listStartTime := time.Now()
	listEndTime := listStartTime.Add(ctlreadTestDuration)

	listProgressDone := make(chan struct{})
	go func() {
		defer close(listProgressDone)
		progressTicker := time.NewTicker(300 * time.Second) // 5 min
		defer progressTicker.Stop()

		for {
			select {
			case <-listCtx.Done():
				return
			case <-progressTicker.C:
				now := time.Now()
				if now.After(listEndTime) {
					return
				}
				elapsed := now.Sub(listStartTime)
				remaining := ctlreadTestDuration - elapsed
				progress := float64(elapsed) / float64(ctlreadTestDuration) * 100

				queryCount := atomic.LoadUint64(&listQueryCounter)

				t.Logf("[LIST] Progress: %.1f%% | Elapsed: %v | Remaining: %v | Queries: %d ",
					progress, elapsed.Round(time.Second), remaining.Round(time.Second), queryCount)
			}
		}
	}()

	listTicker := time.NewTicker(1 * time.Second)
	defer listTicker.Stop()

	for {
		select {
		case <-listCtx.Done():
			goto afterList
		case <-listTicker.C:
			var wg sync.WaitGroup
			wg.Add(concurrentUsers)
			for workerID := 0; workerID < concurrentUsers; workerID++ {
				go func(workerID int) {
					defer wg.Done()
					tenantIndex := rand.Intn(totalTenants)
					tenantID := fmt.Sprintf("pt-%s-%d", randomPrefix, tenantIndex)
					req := message.QueryClustersReq{
						CommonReqInfo: structs.CommonReqInfo{
							Vendor: constants.ProvidedVendor(utils.GetService().Vendor),
							Region: utils.GetService().Region,
						},
						TenantID: tenantID,
					}
					if testBizType != "" {
						req.BizType = testBizType
					}
					request := utils.QueryReq(req)

					requestCtx, requestCancel := context.WithTimeout(context.Background(), rwInterval)
					defer requestCancel()
					startReqTime := time.Now()
					resp, err := perfClient.Do(request.WithContext(requestCtx))
					duration := time.Since(startReqTime)

					var finalStatusCode int
					var finalDuration time.Duration
					if err != nil {
						mu.Lock()
						wrongMap[err.Error()]++
						mu.Unlock()
						if errors.Is(err, context.DeadlineExceeded) {
							finalStatusCode = 999 // timeout
							finalDuration = rwInterval + 100*time.Millisecond
						} else {
							finalStatusCode = -1
							finalDuration = duration
						}
						log.Error("List query failed", zap.Error(err))
					} else {
						finalStatusCode = resp.StatusCode
						io.Copy(io.Discard, resp.Body)
						resp.Body.Close()
						finalDuration = duration
					}
					listStatusStats.Record(finalStatusCode)
					listTimeStats.Record(finalDuration)
					atomic.AddUint64(&listQueryCounter, 1)
				}(workerID)
			}
			wg.Wait()
		}
	}
afterList:
	<-listProgressDone
	totalListQueries := int(atomic.LoadUint64(&listQueryCounter))
	printPerformanceStats(t, "TestNormalReadPerformance", "List Query", listStatusStats, listTimeStats, totalListQueries, wrongMap)
	t.Log("List query phase finished.")

	t.Log("Starting DETAIL query phase...")

	mu.Lock()
	wrongMap = make(map[string]int)
	mu.Unlock()
	detailCtx, detailCancel := context.WithTimeout(context.Background(), ctlreadTestDuration)
	defer detailCancel()

	detailStartTime := time.Now()
	detailEndTime := detailStartTime.Add(ctlreadTestDuration)
	detailProgressDone := make(chan struct{})
	go func() {
		defer close(detailProgressDone)
		progressTicker := time.NewTicker(300 * time.Second) // 5 min
		defer progressTicker.Stop()

		for {
			select {
			case <-detailCtx.Done():
				return
			case <-progressTicker.C:
				now := time.Now()
				if now.After(detailEndTime) {
					return
				}
				elapsed := now.Sub(detailStartTime)
				remaining := ctlreadTestDuration - elapsed
				progress := float64(elapsed) / float64(ctlreadTestDuration) * 100

				queryCount := atomic.LoadUint64(&detailQueryCounter)

				t.Logf("[DETAIL] Progress: %.1f%% | Elapsed: %v | Remaining: %v | Queries: %d",
					progress, elapsed.Round(time.Second), remaining.Round(time.Second), queryCount)
			}
		}
	}()

	detailTicker := time.NewTicker(1 * time.Second)
	defer detailTicker.Stop()

	for {
		select {
		case <-detailCtx.Done():
			goto afterDetail
		case <-detailTicker.C:
			var wg sync.WaitGroup
			wg.Add(concurrentUsers)
			for workerID := 0; workerID < concurrentUsers; workerID++ {
				go func(workerID int) {
					defer wg.Done()
					clusterIndex := rand.Intn(len(createdClusters))
					clusterID := createdClusters[clusterIndex]
					req := message.DetailClusterReq{
						CommonReqInfo: structs.CommonReqInfo{
							Vendor: constants.ProvidedVendor(utils.GetService().Vendor),
							Region: utils.GetService().Region,
						},
						ID: clusterID,
					}
					if testBizType != "" {
						req.BizType = testBizType
					}
					request := utils.DetailReq(req)

					requestCtx, requestCancel := context.WithTimeout(context.Background(), rwInterval)
					startReqTime := time.Now()
					resp, err := perfClient.Do(request.WithContext(requestCtx))
					duration := time.Since(startReqTime)
					defer requestCancel()

					var finalStatusCode int
					var finalDuration time.Duration
					if err != nil {
						mu.Lock()
						wrongMap[err.Error()]++
						mu.Unlock()
						if errors.Is(err, context.DeadlineExceeded) {
							finalStatusCode = 999 //timeout
							finalDuration = rwInterval + 100*time.Millisecond
						} else {
							fmt.Println(err)
							finalStatusCode = -1
							finalDuration = duration
						}
						log.Error("Detail query failed", zap.Error(err))
					} else {
						finalStatusCode = resp.StatusCode
						io.Copy(io.Discard, resp.Body)
						resp.Body.Close()
						finalDuration = duration
					}
					detailStatusStats.Record(finalStatusCode)
					detailTimeStats.Record(finalDuration)
					atomic.AddUint64(&detailQueryCounter, 1)
				}(workerID)
			}
			wg.Wait()
		}
	}
afterDetail:
	<-detailProgressDone
	totalDetailQueries := int(atomic.LoadUint64(&detailQueryCounter))
	printPerformanceStats(t, "TestNormalReadPerformance", "Detail Query", detailStatusStats, detailTimeStats, totalDetailQueries, wrongMap)
}

func TestColdStartReadPerformance(t *testing.T) {
	createdClusters, _ = loadClusterIDs("created_clusters.txt")
	if len(createdClusters) == 0 {
		t.Fatal("No clusters available for read test, please run write test first")
	}

	randomPrefix, err := loadRandomPrefix("created_clusters.txt")
	if err != nil {
		t.Fatalf("Failed to load random prefix: %v", err)
	}
	t.Logf("Loaded random prefix for cold start read test: %s", randomPrefix)

	t.Log("Starting cold start read performance test...")
	f, _ := os.OpenFile(fmt.Sprintf("log/error_TestColdStartReadPerformance_List_%s.log", time.Now().Format("20060102_150405")), os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0666)

	log := getLogger(f)

	mu.Lock()
	wrongMap = make(map[string]int)
	mu.Unlock()

	listStatusStats = &StatusCodeStats{counts: make(map[int]int)}
	listTimeStats = &DurationStats{}
	detailStatusStats = &StatusCodeStats{counts: make(map[int]int)}
	detailTimeStats = &DurationStats{}
	atomic.StoreUint64(&listQueryCounter, 0)
	atomic.StoreUint64(&detailQueryCounter, 0)

	// ==========LIST QUERY PHASE==========
	listCtx, listCancel := context.WithTimeout(context.Background(), ctlreadTestDuration)
	defer listCancel()

	listStartTime := time.Now()
	listEndTime := listStartTime.Add(ctlreadTestDuration)

	listProgressDone := make(chan struct{})
	go func() {
		defer close(listProgressDone)
		progressTicker := time.NewTicker(300 * time.Second) // 5 min
		defer progressTicker.Stop()

		for {
			select {
			case <-listCtx.Done():
				return
			case <-progressTicker.C:
				now := time.Now()
				if now.After(listEndTime) {
					return
				}
				elapsed := now.Sub(listStartTime)
				remaining := ctlreadTestDuration - elapsed
				progress := float64(elapsed) / float64(ctlreadTestDuration) * 100

				queryCount := atomic.LoadUint64(&listQueryCounter)

				t.Logf("[LIST] Progress: %.1f%% | Elapsed: %v | Remaining: %v | Queries: %d ",
					progress, elapsed.Round(time.Second), remaining.Round(time.Second), queryCount)
			}
		}
	}()

	listTicker := time.NewTicker(1 * time.Second)
	defer listTicker.Stop()

	for {
		select {
		case <-listCtx.Done():
			goto afterList
		case <-listTicker.C:
			var wg sync.WaitGroup
			wg.Add(concurrentUsers)
			for workerID := 0; workerID < concurrentUsers; workerID++ {
				go func(workerID int) {
					defer wg.Done()
					s := utils.GetService()
					tenantIndex := rand.Intn(totalTenants)
					tenantID := fmt.Sprintf("pt-%s-%d", randomPrefix, tenantIndex)
					req := message.QueryClustersReq{
						CommonReqInfo: structs.CommonReqInfo{
							Vendor: constants.ProvidedVendor(s.Vendor),
							Region: s.Region,
						},
						TenantID: tenantID,
					}
					request := utils.QueryReq(req)

					requestCtx, requestCancel := context.WithTimeout(context.Background(), rwInterval)
					defer requestCancel()

					startReqTime := time.Now()
					resp, err := perfClient.Do(request.WithContext(requestCtx))
					duration := time.Since(startReqTime)

					var finalStatusCode int
					var finalDuration time.Duration
					if err != nil {
						mu.Lock()
						wrongMap[err.Error()]++
						mu.Unlock()
						if errors.Is(err, context.DeadlineExceeded) {
							finalStatusCode = 999 // timeout
							finalDuration = rwInterval + 100*time.Millisecond
						} else {
							finalStatusCode = -1
							finalDuration = duration
						}
						log.Error("List query failed", zap.Error(err))
					} else {
						finalStatusCode = resp.StatusCode
						io.Copy(io.Discard, resp.Body)
						resp.Body.Close()
						finalDuration = duration
					}

					listStatusStats.Record(finalStatusCode)
					listTimeStats.Record(finalDuration)
					atomic.AddUint64(&listQueryCounter, 1)
				}(workerID)
			}
			wg.Wait()
		}
	}
afterList:
	<-listProgressDone
	totalListQueries := int(atomic.LoadUint64(&listQueryCounter))
	printPerformanceStats(t, "TestColdStartReadPerformance", "List Query", listStatusStats, listTimeStats, totalListQueries, wrongMap)
	t.Log("List query phase finished.")

	// ==========DETAIL QUERY PHASE==========
	t.Log("Starting DETAIL query phase...")
	mu.Lock()
	wrongMap = make(map[string]int)
	mu.Unlock()
	detailCtx, detailCancel := context.WithTimeout(context.Background(), ctlreadTestDuration)
	defer detailCancel()

	detailStartTime := time.Now()
	detailEndTime := detailStartTime.Add(ctlreadTestDuration)

	detailProgressDone := make(chan struct{})
	go func() {
		defer close(detailProgressDone)
		progressTicker := time.NewTicker(300 * time.Second)
		defer progressTicker.Stop()

		for {
			select {
			case <-detailCtx.Done():
				return
			case <-progressTicker.C:
				now := time.Now()
				if now.After(detailEndTime) {
					return
				}
				elapsed := now.Sub(detailStartTime)
				remaining := ctlreadTestDuration - elapsed
				progress := float64(elapsed) / float64(ctlreadTestDuration) * 100

				queryCount := atomic.LoadUint64(&detailQueryCounter)

				t.Logf("[DETAIL] Progress: %.1f%% | Elapsed: %v | Remaining: %v | Queries: %d",
					progress, elapsed.Round(time.Second), remaining.Round(time.Second), queryCount)
			}
		}
	}()

	detailTicker := time.NewTicker(1 * time.Second)
	defer detailTicker.Stop()

	for {
		select {
		case <-detailCtx.Done():
			goto afterDetail
		case <-detailTicker.C:
			var wg sync.WaitGroup
			wg.Add(concurrentUsers)
			for workerID := 0; workerID < concurrentUsers; workerID++ {
				go func(workerID int) {
					defer wg.Done()
					s := utils.GetService()

					createdClustersMu.Lock()
					clusterIndex := rand.Intn(len(createdClusters))
					clusterID := createdClusters[clusterIndex]
					createdClustersMu.Unlock()

					req := message.DetailClusterReq{
						CommonReqInfo: structs.CommonReqInfo{
							Vendor: constants.ProvidedVendor(s.Vendor),
							Region: s.Region,
						},
						ID: clusterID,
					}
					request := utils.DetailReq(req)

					requestCtx, requestCancel := context.WithTimeout(context.Background(), rwInterval)
					defer requestCancel()
					startReqTime := time.Now()
					resp, err := perfClient.Do(request.WithContext(requestCtx))
					duration := time.Since(startReqTime)

					var finalStatusCode int
					var finalDuration time.Duration
					if err != nil {
						mu.Lock()
						wrongMap[err.Error()]++
						mu.Unlock()
						if errors.Is(err, context.DeadlineExceeded) {
							finalStatusCode = 999 // timeout
							finalDuration = rwInterval + 100*time.Millisecond
						} else {
							finalStatusCode = -1
							finalDuration = duration
						}
						log.Error("Detail query failed", zap.Error(err))
					} else {
						finalStatusCode = resp.StatusCode
						io.Copy(io.Discard, resp.Body)
						resp.Body.Close()
						finalDuration = duration
					}

					detailStatusStats.Record(finalStatusCode)
					detailTimeStats.Record(finalDuration)
					atomic.AddUint64(&detailQueryCounter, 1)
				}(workerID)
			}
			wg.Wait()
		}
	}
afterDetail:
	<-detailProgressDone
	totalDetailQueries := int(atomic.LoadUint64(&detailQueryCounter))
	printPerformanceStats(t, "TestColdStartReadPerformance", "Detail Query", detailStatusStats, detailTimeStats, totalDetailQueries, wrongMap)
}

func TestDataPlaneWritePerformance(t *testing.T) {

	// clear wrongMap
	mu.Lock()
	wrongMap = make(map[string]int)
	mu.Unlock()

	t.Logf("Starting data plane write performance test: totalClusters=%d, metricsPerWrite=%d, writeInterval=%v, duration=%v",
		totalClusters, metricsPerWrite, writeInterval, writeTestDuration)
	f, _ := os.OpenFile(fmt.Sprintf("log/error_TestDataPlaneWritePerformance_%s.log", time.Now().Format("20060102_150405")), os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0666)

	log := getLogger(f)

	ctx, cancel := context.WithTimeout(context.Background(), writeTestDuration)
	defer cancel()

	clusterIDs := make([]string, totalClusters)
	for i := 0; i < totalClusters; i++ {
		clusterIDs[i] = fmt.Sprintf("dataplane-cluster-%d", i)
	}

	startTime := time.Now()
	// print progress
	endTime := startTime.Add(writeTestDuration)
	progressDone := make(chan struct{})
	go func() {
		defer close(progressDone)
		progressTicker := time.NewTicker(300 * time.Second) //5 min
		defer progressTicker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-progressTicker.C:
				now := time.Now()
				if now.After(endTime) {
					return
				}
				elapsed := now.Sub(startTime)
				remaining := writeTestDuration - elapsed
				progress := float64(elapsed) / float64(writeTestDuration) * 100

				writeCount := atomic.LoadUint64(&writeCounter)

				t.Logf("Progress: %.1f%% | Elapsed: %v | Remaining: %v | Writes: %d",
					progress, elapsed.Round(time.Second), remaining.Round(time.Second), writeCount)
			}
		}
	}()

	ticker := time.NewTicker(writeInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			goto done
		case <-ticker.C:
			var wg sync.WaitGroup
			wg.Add(len(clusterIDs))
			for _, clusterID := range clusterIDs {
				go func(clusterID string) {
					defer wg.Done()
					writeReq := generateWriteRequest(clusterID)
					requestCtx, requestCancel := context.WithTimeout(context.Background(), writeInterval)
					defer requestCancel()
					startReqTime := time.Now()
					resp, err := writeMetrics(requestCtx, writeReq)
					duration := time.Since(startReqTime)

					var statusCode int
					if err != nil {
						mu.Lock()
						wrongMap[err.Error()]++
						mu.Unlock()
						if errors.Is(err, context.DeadlineExceeded) {
							statusCode = 999 //timeout
							writeTimeStats.Record(writeInterval + 100*time.Millisecond)
						} else {
							statusCode = -1
							writeTimeStats.Record(duration)
						}
						log.Error("error", zap.Error(err))
					} else {
						statusCode = resp.StatusCode
						resp.Body.Close()
						writeTimeStats.Record(duration)
					}
					writeStatusStats.Record(statusCode)
					atomic.AddUint64(&writeCounter, 1)
				}(clusterID)
			}
			wg.Wait()
		}
	}

done:
	<-progressDone
	totalWrites := int(atomic.LoadUint64(&writeCounter))
	totalTestTime := time.Since(startTime)

	printPerformanceStats(t, "TestDataPlaneWritePerformance", "Metrics Write", writeStatusStats, writeTimeStats, totalWrites, wrongMap)
	t.Logf("Total write test time: %v", totalTestTime)

	assert.Greater(t, totalWrites, 0, "Should successfully complete at least some write operations")
}

func TestDataPlaneReadPerformance(t *testing.T) {
	// clear wrongMap
	mu.Lock()
	wrongMap = make(map[string]int)
	mu.Unlock()

	readStatusStats = &StatusCodeStats{counts: make(map[int]int)}
	readTimeStats = &DurationStats{}
	atomic.StoreUint64(&readCounter, 0)

	t.Logf("Starting cold start read performance test: totalTenants=%d, metricsCount=%d, testDuration=%v",
		totalDataTenants, readMetricsCount, dataReadTestDuration)
	f, _ := os.OpenFile(fmt.Sprintf("log/error_TestDataPlaneReadPerformance_%s.log", time.Now().Format("20060102_150405")), os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0666)

	log := getLogger(f)

	ctx, cancel := context.WithTimeout(context.Background(), dataReadTestDuration)
	defer cancel()

	clusterIDs := make([]string, totalDataTenants)
	for i := 0; i < totalDataTenants; i++ {
		clusterIDs[i] = fmt.Sprintf("dataplane-cluster-%d", i)
	}

	startTime := time.Now()
	// print progress
	endTime := startTime.Add(dataReadTestDuration)
	progressDone := make(chan struct{})
	go func() {
		defer close(progressDone)
		progressTicker := time.NewTicker(300 * time.Second) // 5 min
		defer progressTicker.Stop()

		for {
			select {
			case <-ctx.Done():
				return
			case <-progressTicker.C:
				now := time.Now()
				if now.After(endTime) {
					return
				}
				elapsed := now.Sub(startTime)
				remaining := dataReadTestDuration - elapsed
				progress := float64(elapsed) / float64(dataReadTestDuration) * 100

				requestCount := atomic.LoadUint64(&readCounter)

				t.Logf("Progress: %.1f%% | Elapsed: %v | Remaining: %v | Requests: %d",
					progress, elapsed.Round(time.Second), remaining.Round(time.Second), requestCount)
			}
		}
	}()

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

	for {
		select {
		case <-ctx.Done():
			goto done
		case <-ticker.C:
			var outerWG sync.WaitGroup
			outerWG.Add(totalDataTenants)
			for userID := 0; userID < totalDataTenants; userID++ {
				go func(userID int) {
					defer outerWG.Done()
					var innerWG sync.WaitGroup
					innerWG.Add(readMetricsCount)
					for metricIndex := 0; metricIndex < readMetricsCount; metricIndex++ {
						go func(metricIndex int) {
							defer innerWG.Done()
							idx := rand.Intn(len(clusterIDs))
							clusterID := clusterIDs[idx]

							readReq := generateReadRequest(clusterID)
							readReq.MetricNames = []string{fmt.Sprintf("test_metric_%d", metricIndex)}

							requestCtx, requestCancel := context.WithTimeout(context.Background(), readInterval)
							defer requestCancel()
							startReqTime := time.Now()
							resp, err := readMetrics(requestCtx, readReq)
							duration := time.Since(startReqTime)

							var finalStatusCode int
							if err != nil {
								mu.Lock()
								wrongMap[err.Error()]++
								mu.Unlock()
								if errors.Is(err, context.DeadlineExceeded) {
									finalStatusCode = 999 //timeout
									readTimeStats.Record(readInterval + 100*time.Millisecond)
								} else {
									finalStatusCode = -1
									readTimeStats.Record(duration)
								}
								log.Error("error", zap.Error(err))
							} else {
								finalStatusCode = resp.StatusCode
								io.Copy(io.Discard, resp.Body)
								resp.Body.Close()
								readTimeStats.Record(duration)
							}
							readStatusStats.Record(finalStatusCode)
							atomic.AddUint64(&readCounter, 1)
						}(metricIndex)
					}
					innerWG.Wait()
				}(userID)
			}
			outerWG.Wait()
		}
	}

done:
	<-progressDone
	totalQueries := int(atomic.LoadUint64(&readCounter))
	totalTestTime := time.Since(startTime)
	printPerformanceStats(t, "TestDataPlaneReadPerformance", "Metrics Read", readStatusStats, readTimeStats, totalQueries, wrongMap)
	t.Logf("Total read test time: %v", totalTestTime)
}

func TestDataPlaneConsoleReadPerformance(t *testing.T) {
	// clear wrongMap
	mu.Lock()
	wrongMap = make(map[string]int)
	mu.Unlock()

	readStatusStats = &StatusCodeStats{counts: make(map[int]int)}
	readTimeStats = &DurationStats{}
	atomic.StoreUint64(&readCounter, 0)

	t.Logf("Starting cold start read performance test: totalTenants=%d, metricsCount=%d, testDuration=%v",
		totalDataTenants, readMetricsCount, dataReadTestDuration)
	f, _ := os.OpenFile(fmt.Sprintf("log/error_TestDataPlaneReadPerformance_%s.log", time.Now().Format("20060102_150405")), os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0666)

	log := getLogger(f)

	ctx, cancel := context.WithTimeout(context.Background(), dataReadTestDuration)
	defer cancel()

	clusterIDs := make([]string, totalDataTenants)
	for i := 0; i < totalDataTenants; i++ {
		clusterIDs[i] = fmt.Sprintf("dataplane-cluster-%d", i)
	}

	startTime := time.Now()
	// print progress
	endTime := startTime.Add(dataReadTestDuration)
	progressDone := make(chan struct{})
	go func() {
		defer close(progressDone)
		progressTicker := time.NewTicker(300 * time.Second) // 5 min
		defer progressTicker.Stop()

		for {
			select {
			case <-ctx.Done():
				return
			case <-progressTicker.C:
				now := time.Now()
				if now.After(endTime) {
					return
				}
				elapsed := now.Sub(startTime)
				remaining := dataReadTestDuration - elapsed
				progress := float64(elapsed) / float64(dataReadTestDuration) * 100

				requestCount := atomic.LoadUint64(&readCounter)

				t.Logf("Progress: %.1f%% | Elapsed: %v | Remaining: %v | Requests: %d",
					progress, elapsed.Round(time.Second), remaining.Round(time.Second), requestCount)
			}
		}
	}()

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

	for {
		select {
		case <-ctx.Done():
			goto done
		case <-ticker.C:
			var outerWG sync.WaitGroup
			outerWG.Add(totalDataTenants)
			for userID := 0; userID < totalDataTenants; userID++ {
				go func(userID int) {
					defer outerWG.Done()
					var innerWG sync.WaitGroup

					batchSize := 8
					batchCount := (readMetricsCount + batchSize - 1) / batchSize

					for batch := 0; batch < batchCount; batch++ {
						start := batch * batchSize
						end := start + batchSize
						if end > readMetricsCount {
							end = readMetricsCount
						}
						currentBatchSize := end - start
						innerWG.Add(currentBatchSize)

						for metricIndex := start; metricIndex < end; metricIndex++ {
							go func(metricIndex int) {
								defer innerWG.Done()
								idx := rand.Intn(len(clusterIDs))
								clusterID := clusterIDs[idx]

								readReq := generateReadRequest(clusterID)
								readReq.MetricNames = []string{fmt.Sprintf("test_metric_%d", metricIndex)}

								requestCtx, requestCancel := context.WithTimeout(context.Background(), readInterval)
								defer requestCancel()
								startReqTime := time.Now()
								resp, err := readMetrics(requestCtx, readReq)
								duration := time.Since(startReqTime)

								var finalStatusCode int
								if err != nil {
									mu.Lock()
									wrongMap[err.Error()]++
									mu.Unlock()
									if errors.Is(err, context.DeadlineExceeded) {
										finalStatusCode = 999 //timeout
										readTimeStats.Record(readInterval + 100*time.Millisecond)
									} else {
										finalStatusCode = -1
										readTimeStats.Record(duration)
									}
									log.Error("error", zap.Error(err))
								} else {
									finalStatusCode = resp.StatusCode
									io.Copy(io.Discard, resp.Body)
									resp.Body.Close()
									readTimeStats.Record(duration)
								}
								readStatusStats.Record(finalStatusCode)
								atomic.AddUint64(&readCounter, 1)
							}(metricIndex)
						}

						innerWG.Wait()
					}
				}(userID)
			}
			outerWG.Wait()
		}
	}

done:
	<-progressDone
	totalQueries := int(atomic.LoadUint64(&readCounter))
	totalTestTime := time.Since(startTime)
	printPerformanceStats(t, "TestDataPlaneReadPerformance", "Metrics Read", readStatusStats, readTimeStats, totalQueries, wrongMap)
	t.Logf("Total read test time: %v", totalTestTime)
}

func writeMetrics(ctx context.Context, req WriteMetricsRequest) (*http.Response, error) {
	//apiURL := fmt.Sprintf("http://%s:%d/insert/92362:0/prometheus/api/v1/import", vmInsertLocalHost, vmInsertLocalPort)
	apiURL := fmt.Sprintf("http://nlb-slmuwp82g2dypwskva.ap-southeast-1.nlb.aliyuncsslbintl.com/76fcbc66-1b24-478e-a13c-04a249f2b239/api/v1/import")
	var jsonPayloads []string

	for _, metric := range req.Metrics {
		for _, point := range metric.DataPoints {
			metricObj := map[string]string{
				"__name__": metric.Name,
			}
			for k, v := range point.Labels {
				metricObj[k] = v
			}

			payload := map[string]interface{}{
				"metric":     metricObj,
				"timestamps": []int64{point.Timestamp * 1000},
				"values":     []float64{point.Value},
			}

			// Marshal to JSON
			jsonData, err := json.Marshal(payload)
			if err != nil {
				return nil, fmt.Errorf("error marshaling payload: %v", err)
			}
			//fmt.Println(string(jsonData))
			jsonPayloads = append(jsonPayloads, string(jsonData))
		}
	}
	//fmt.Println(jsonPayloads)
	jsonLines := strings.Join(jsonPayloads, "\n")
	//fmt.Println(jsonLines)
	request, err := http.NewRequestWithContext(ctx, "POST", apiURL, strings.NewReader(jsonLines))
	if err != nil {
		return nil, fmt.Errorf("error creating write request: %v", err)
	}
	request.Header.Set("Content-Type", "application/json")

	return perfClient.Do(request)
}

// read metrics using port-forward
func readMetrics(ctx context.Context, req ReadMetricsRequest) (*http.Response, error) {
	//baseURL := fmt.Sprintf("http://%s:%d/select/92362:0/prometheus/api/v1/query_range", vmSelectLocalHost, vmSelectLocalPort)
	baseURL := fmt.Sprintf("https://www.ds.ap-southeast-1.alicloud.observability-dev.pingcap.cloud/internal/metrics/76fcbc66-1b24-478e-a13c-04a249f2b239/api/v1/query_range")

	params := url.Values{}

	if len(req.MetricNames) > 0 {
		query := fmt.Sprintf("%s{cluster_id=\"%s\"}", req.MetricNames[0], req.ClusterID)
		params.Add("query", query)
	} else {
		params.Add("query", "up")
	}

	// For range query, add start, end and step parameters
	params.Add("start", fmt.Sprintf("%d", req.StartTime))
	params.Add("end", fmt.Sprintf("%d", req.EndTime))
	params.Add("step", fmt.Sprintf("%d", req.Step))

	request, err := http.NewRequestWithContext(ctx, "GET", baseURL+"?"+params.Encode(), nil)
	if err != nil {
		return nil, fmt.Errorf("error creating read request: %v", err)
	}
	request.Header.Set("Content-Type", "application/json")

	return perfClient.Do(request)
}

func TestDeleteRegionalPod(t *testing.T) {
	labelSelector := "app=regionalserver"
	nameSpace := "observability"
	utils.DeletePod(nameSpace, labelSelector)
	t.Log("Regional Server delete succ")
}

func TestDeleteVMPod(t *testing.T) {
	labelSelector1 := "app.kubernetes.io/name=vmselect"
	labelSelector2 := "app.kubernetes.io/name=vmstorage"
	nameSpace := "observability"
	utils.DeletePod(nameSpace, labelSelector1)
	utils.DeletePod(nameSpace, labelSelector2)
	t.Log("VM Cluster delete succ")
}

func TestClearResource(t *testing.T) {
	log := getLogger(os.Stdout)
	log.Info("Cleaning up historical resources...")
	utils.ClearNamespacesForPerformanceTest()
}