package ccb_application_router

import (
	"context"
	"fmt"
	"strings"
	"time"

	. "github.com/onsi/ginkgo"
	v1 "github.com/prometheus/client_golang/api/prometheus/v1"
	"golang.org/x/sync/errgroup"

	. "github.com/onsi/gomega"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"

	pdclient "github.com/pingcap/endless/pkg/client/pd"
	. "github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/pkg/workloadnode"

	_ "github.com/pingcap/endless/pkg/client/tidb"
)

var TESTCASE_Stage_1_NAME = "Stage 1 Empty Cluster-Client-6.6K TPS"
var TESTCASE_1_PROMQL = []CheckPromQL{
	CheckPromQL{name: "rawkvworkload.err",
		query: "sum(rawkvworkload_trx_failures_total)",
		step:  1 * time.Second,
		fn:    MetricsDelta,
	},
	CheckPromQL{name: "rawkvworkload.lat.avg",
		query: "sum(idelta(rawkvworkload_trx_latency_seconds_sum[60s]))/sum(idelta(rawkvworkload_trx_latency_seconds_count[60s]))",
		step:  1 * time.Second,
		fn:    MetricsMax,
	},
	CheckPromQL{name: "rawkvworkload.lat.p99",
		query: "histogram_quantile(0.99, sum(rate(rawkvworkload_trx_latency_seconds_bucket[60s])) by (le))",
		step:  1 * time.Second,
		fn:    MetricsMax,
	},
	CheckPromQL{name: "rawkvworkload.tps",
		query: "sum(rate(rawkvworkload_trx_latency_seconds_count[60s]))",
		step:  1 * time.Second,
		fn:    MetricsMax,
	},
	// CheckPromQL{name: "client.qps",
	// 	query: "sum(rate(client_java_grpc_raw_requests_latency_count[60s]))",
	// 	step:  1 * time.Second,
	// 	fn:    MetricsMax,
	// },
	CheckPromQL{name: "client.lat.avg",
		query: "sum(idelta(client_java_raw_requests_latency_sum[60s]))/sum(idelta(client_java_raw_requests_latency_count[60s]))",
		step:  1 * time.Second,
		fn:    MetricsMax,
	},
	CheckPromQL{name: "client.lat.p99",
		query: "histogram_quantile(0.99, sum(rate(client_java_raw_requests_latency_bucket[60s])) by (le))",
		step:  1 * time.Second,
		fn:    MetricsMax,
	},
	CheckPromQL{name: "client.err",
		query: "sum(client_java_raw_requests_failure_total) or vector(0)",
		step:  1 * time.Second,
		fn:    MetricsDelta,
	},
}

type TestCaseStageInfo struct {
	Start       time.Time
	End         time.Time
	Description string
}

func GenerateReport(ctx context.Context, workloadClient resource.WorkloadNode, StagesTSInfo []TestCaseStageInfo) {
	var reportConfigInfo strings.Builder
	reportConfigInfo.WriteString("summary,start,end,\n")
	for _, stageInfo := range StagesTSInfo {
		reportConfigInfo.WriteString(stageInfo.Description + ",")
		reportConfigInfo.WriteString(fmt.Sprintf("%v,", stageInfo.Start.Unix()*1000))
		reportConfigInfo.WriteString(fmt.Sprintf("%v,\n", stageInfo.End.Unix()*1000))
	}
	fileName := fmt.Sprintf("test.%s.periods.csv", time.Now().Format("2006-01-02"))
	cmd := fmt.Sprintf("echo \"%s\" > /usr/src/rawkvworkload/report_clj/%s", strings.Trim(reportConfigInfo.String(), "\n"), fileName)
	err := workloadnode.CommandContext(ctx, cmd).Run(workloadClient)
	Expect(err).Should(BeNil())
	// Todo: pd version, tikv version, client-version
	cmd = fmt.Sprintf("cd /usr/src/rawkvworkload/report_clj/; bb -m report 1 2 3 %s", fileName)
	err = workloadnode.CommandContext(ctx, cmd).Run(workloadClient)
	log.S().Infof("Generate Report by BB with Err: %v", err)
}

func StageLast(stageDesc string, dur time.Duration) TestCaseStageInfo {
	stage := TestCaseStageInfo{
		Description: stageDesc,
		Start:       time.Now(),
	}
	time.Sleep(dur)
	stage.End = time.Now()
	return stage
}

func CCBLaunchWithImportData(suiteTestCtx core.TestContext, ctx context.Context, tc resource.TiDBCluster, ccbloadClientAPIPrometheus v1.API, APIVersion int) {
	pdClient := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	pdClientAddr := pdClient.GetPDAddr()
	wn := suiteTestCtx.Resource("workload0").(resource.WorkloadNode)
	SetupPDWithPDCTL(pdClientAddr, wn, ctx)
	log.Info("Update client info into ccbload clientmonitor Prometheus Config")
	UpdatePrometheusConfig(suiteTestCtx, ctx, 15)
	g, gctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		// log.S().Infof("Update client info into ccbload clientmonitor Prometheus Config, %v", gctx)
		// return nil
		return ImporterClientPrepareData(suiteTestCtx, gctx)
	})
	StopCCBLoadClient(wn, 0, 5)
	testcaseTimestamp := make([]TestCaseStageInfo, 0)
	By("At stage 1, launching some clients to hit TPS 6.6k")
	By("At stage 1-1, set read-only")
	SetImporterRW(suiteTestCtx, ctx, pdClientAddr, "onlyread")
	log.Info("Testcase runs")
	By("At stage 1-2, start workloads to supply stress(TPS: 6.6k)")
	log.Info("At stage 1-2, start workloads to supply stress(TPS: 6.6k)")
	LaunchCCBLoadClient(pdClientAddr, wn, 0, 2, APIVersion, 5*3600, 15, ctx)
	// TODO: adjust the sleep duration before running on prod
	stageInfo := StageLast("Stage 1 Empty Cluster-Client-TPS 6.6K", 30*time.Minute)
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)

	By("At stage 2, start workloads to supply stress(TPS: 8.8k)")
	LaunchCCBLoadClient(pdClientAddr, wn, 2, 1, APIVersion, 5*3600, 20, ctx)
	// TODO: adjust the sleep duration before running on prod
	stageInfo = StageLast("Stage 2 Empty Cluster-Client-TPS 8.8k)", 30*time.Minute)
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)

	By("At stage 3, start workloads to supply stress(TPS: 22k)")
	LaunchCCBLoadClient(pdClientAddr, wn, 3, 3, APIVersion, 5*3600, 25, ctx)
	// time.Sleep(30 * time.Minute)
	stageInfo = StageLast("Stage 3 Empty Cluster-Client-TPS 22k)", 30*time.Minute)
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)
	// Todo: check TPS, latency, slowlog
	// wait tar xvf done, then import the data into the cluster
	Expect(g.Wait()).Should(BeNil())
	By("At stage 4, import data from 2 banks")
	stageInfo = ImportData(suiteTestCtx, ctx, "importer.properties", pdClientAddr, 1, "Stage 4 Import data of 2 banks")
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)
	time.Sleep(5 * time.Minute)
	By("At stage 5, delete data")
	stageInfo = ImportData(suiteTestCtx, ctx, "importer-delete.properties", pdClientAddr, 2, "Stage 5 Delete data")
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)
	time.Sleep(5 * time.Minute)
	By("At stage 6, import data from 2 banks again")
	ImportData(suiteTestCtx, ctx, "importer.properties", pdClientAddr, 3, "Stage 6 Import data of 2 banks again")
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)
	time.Sleep(5 * time.Minute)
	By("At stage 7, R-W run the workload")
	// set state
	ImportData(suiteTestCtx, ctx, "importer-setstate.properties", pdClientAddr, 4, "Stage 7 set state")
	SetImporterRW(suiteTestCtx, ctx, pdClientAddr, "r-w")
	stageInfo = StageLast("Stage 7 R-W run the workload", 60*time.Minute)
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)
	GenerateReport(ctx, wn, testcaseTimestamp)
	VerifyEachStageMetrics(ctx, ccbloadClientAPIPrometheus, TESTCASE_1_PROMQL, testcaseTimestamp)
}

func CCBLaunchWithRestoreData(suiteTestCtx core.TestContext, ctx context.Context, tc resource.TiDBCluster, ccbloadClientAPIPrometheus v1.API, APIVersion int, RestorePathOnMinio string) {
	pdClient := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	pdClientAddr := pdClient.GetPDAddr()
	wn := suiteTestCtx.Resource("workload0").(resource.WorkloadNode)
	SetupPDWithPDCTL(pdClientAddr, wn, ctx)
	log.Info("Update client info into ccbload clientmonitor Prometheus Config")
	UpdatePrometheusConfig(suiteTestCtx, ctx, 15)
	StopCCBLoadClient(wn, 0, 5)
	testcaseTimestamp := make([]TestCaseStageInfo, 0)
	tiupWorkloadNode := suiteTestCtx.Resource("tiup").(resource.WorkloadNode)
	RawKVRestore("Restore 1e kv data", tiupWorkloadNode, pdClientAddr, RestorePathOnMinio)

	By("At stage 1, launching some clients to hit TPS 6.6k")
	log.Info("Testcase runs")
	log.Info("At stage 1, start workloads to supply stress(TPS: 6.6k)")
	LaunchCCBLoadClient(pdClientAddr, wn, 0, 2, APIVersion, 0, 15, ctx)
	// TODO: adjust the sleep duration before running on prod
	stageInfo := StageLast("Stage 1 Client-TPS 6.6K", 30*time.Minute)
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)

	By("At stage 2, start workloads to supply stress(TPS: 8.8k)")
	LaunchCCBLoadClient(pdClientAddr, wn, 2, 1, APIVersion, 0, 20, ctx)
	// TODO: adjust the sleep duration before running on prod
	stageInfo = StageLast("Stage 2 Client-TPS 8.8k)", 30*time.Minute)
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)

	By("At stage 3, start workloads to supply stress(TPS: 22k)")
	LaunchCCBLoadClient(pdClientAddr, wn, 3, 3, APIVersion, 0, 25, ctx)
	// time.Sleep(30 * time.Minute)
	stageInfo = StageLast("Stage 3 Cluster-Client-TPS 22k)", 30*time.Minute)
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)
	// Todo: check TPS, latency, slowlog
	time.Sleep(7 * 60 * time.Minute)
	stageInfo = StageLast("Stage 4 R-W run the workload", 60*time.Minute)
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)
	GenerateReport(ctx, wn, testcaseTimestamp)
	VerifyEachStageMetrics(ctx, ccbloadClientAPIPrometheus, TESTCASE_1_PROMQL, testcaseTimestamp)
}

func CCBLaunchWithBatchPutData(suiteTestCtx core.TestContext, ctx context.Context, tc resource.TiDBCluster, ccbloadClientAPIPrometheus v1.API, APIVersion int) {
	pdClient := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	pdClientAddr := pdClient.GetPDAddr()
	wn := suiteTestCtx.Resource("workload0").(resource.WorkloadNode)
	SetupPDWithPDCTL(pdClientAddr, wn, ctx)
	log.Info("Update client info into ccbload clientmonitor Prometheus Config")
	UpdatePrometheusConfig(suiteTestCtx, ctx, 15)
	g, _ := ImportDataByBatchPut(pdClientAddr, wn, 0, 6, APIVersion, 6*1e6, ctx)
	Expect(g.Wait()).Should(BeNil())
	testcaseTimestamp := make([]TestCaseStageInfo, 0)
	log.Info("Testcase runs")
	By("At stage 1, launching some clients to hit TPS 6.6k")
	log.Info("At stage 1, start workloads to supply stress(TPS: 6.6k)")
	LaunchCCBLoadClient(pdClientAddr, wn, 0, 2, APIVersion, 0, 15, ctx)
	// TODO: adjust the sleep duration before running on prod
	stageInfo := StageLast("Stage 1 Cluster-Client-TPS 6.6K", 30*time.Minute)
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)

	By("At stage 2, start workloads to supply stress(TPS: 8.8k)")
	LaunchCCBLoadClient(pdClientAddr, wn, 2, 1, APIVersion, 0, 20, ctx)
	stageInfo = StageLast("Stage 2 Client-TPS 8.8k)", 30*time.Minute)
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)

	By("At stage 3, start workloads to supply stress(TPS: 22k)")
	LaunchCCBLoadClient(pdClientAddr, wn, 3, 3, APIVersion, 5*3600, 25, ctx)
	stageInfo = StageLast("Stage 3 Client-TPS 22k)", 30*time.Minute)
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)

	stageInfo = StageLast("Stage 4 run the workload 1 hour", 60*time.Minute)
	testcaseTimestamp = append(testcaseTimestamp, stageInfo)
	GenerateReport(ctx, wn, testcaseTimestamp)
	VerifyEachStageMetrics(ctx, ccbloadClientAPIPrometheus, TESTCASE_1_PROMQL, testcaseTimestamp)
}