package central_thanos

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

	"github.com/tidbcloud/metrics-integration/pipeline/central/entity"
	"github.com/tidbcloud/metrics-integration/pipeline/internal/datasource"
	"github.com/tidbcloud/metrics-integration/pipeline/internal/datasource/client"
)

var nextGenKeyspaceIDCache sync.Map
var nextGenKeyspaceIDInflight sync.Map

type nextGenKeyspaceLookup struct {
	done       chan struct{}
	keyspaceID string
	err        error
}

const nextGenKeyspaceLookupWindow = "24h"

func aggregateDBMatcherForServicePlan(servicePlan string, clusterID uint64) string {
	if entity.ServicePlan(servicePlan).UsesPremiumMetrics() {
		return fmt.Sprintf(`tidb_cluster_id="%d"`, clusterID)
	}
	switch normalizedServicePlan(servicePlan) {
	case strings.ToLower(string(entity.ServicePlanEssential)),
		strings.ToLower(string(entity.ServicePlanStarter)):
		return fmt.Sprintf(`cluster_id="%d"`, clusterID)
	}

	// Dedicated keeps the historical regex matcher for DB aggregate metrics.
	return fmt.Sprintf(`cluster_id=~".*%d"`, clusterID)
}

func exactClusterMatcherForServicePlan(servicePlan string, clusterID uint64) string {
	if entity.ServicePlan(servicePlan).UsesPremiumMetrics() {
		return fmt.Sprintf(`tidb_cluster_id="%d"`, clusterID)
	}
	switch normalizedServicePlan(servicePlan) {
	case strings.ToLower(string(entity.ServicePlanEssential)),
		strings.ToLower(string(entity.ServicePlanStarter)):
		return fmt.Sprintf(`cluster_id="%d"`, clusterID)
	}

	return fmt.Sprintf(`cluster_id="tidb%d"`, clusterID)
}

func regexClusterMatcherForServicePlan(servicePlan string, clusterID uint64) string {
	if entity.ServicePlan(servicePlan).UsesPremiumMetrics() {
		return fmt.Sprintf(`tidb_cluster_id=~".*%d"`, clusterID)
	}
	switch normalizedServicePlan(servicePlan) {
	case strings.ToLower(string(entity.ServicePlanEssential)),
		strings.ToLower(string(entity.ServicePlanStarter)):
		return fmt.Sprintf(`cluster_id=~".*%d"`, clusterID)
	}

	return fmt.Sprintf(`cluster_id=~".*%d"`, clusterID)
}

func tidbClusterRegexMatcherForServicePlan(servicePlan string, clusterID uint64) string {
	switch normalizedServicePlan(servicePlan) {
	case strings.ToLower(string(entity.ServicePlanEssential)),
		strings.ToLower(string(entity.ServicePlanStarter)):
		return fmt.Sprintf(`cluster_id=~".*%d"`, clusterID)
	default:
		return fmt.Sprintf(`tidb_cluster_id=~".*%d"`, clusterID)
	}
}

func normalizedServicePlan(servicePlan string) string {
	return strings.ToLower(strings.TrimSpace(servicePlan))
}

func servicePlanFromContext(ctx context.Context) entity.ServicePlan {
	addr, err := datasource.GetMetricDataAddressFromContext(ctx)
	if err != nil {
		return entity.ServicePlan("")
	}

	return entity.ServicePlan(addr.TidbClusterInfo.ServicePlan)
}

func sharedSeriesMatcherForServicePlan(servicePlan string, clusterID uint64, keyspaceID string) (string, error) {
	if entity.ServicePlan(servicePlan).UsesPremiumMetrics() {
		return fmt.Sprintf(`tidb_cluster_id="%d"`, clusterID), nil
	}
	switch normalizedServicePlan(servicePlan) {
	case strings.ToLower(string(entity.ServicePlanEssential)),
		strings.ToLower(string(entity.ServicePlanStarter)):
		if keyspaceID == "" {
			return "", fmt.Errorf("keyspace id is empty for service plan %s", servicePlan)
		}
		return fmt.Sprintf(`keyspace_id="%s"`, keyspaceID), nil
	default:
		// Dedicated keeps the historical regex matcher for shared CDC series.
		return regexClusterMatcherForServicePlan(servicePlan, clusterID), nil
	}
}

func buildNextGenKeyspaceLookupPromQL(clusterMatcher string) string {
	return fmt.Sprintf(
		`topk(1, tidb_server_connections{%[1]s}) or topk(1, tidb_server_tokens{%[1]s}) or topk(1, tidb_server_handle_query_duration_seconds_sum{%[1]s})`,
		clusterMatcher,
	)
}

func buildNextGenHistoricalKeyspaceLookupPromQL(clusterMatcher string) string {
	return fmt.Sprintf(
		`topk(1, last_over_time(tidb_server_connections{%[1]s}[%[2]s])) or topk(1, last_over_time(tidb_server_tokens{%[1]s}[%[2]s])) or topk(1, last_over_time(tidb_server_handle_query_duration_seconds_sum{%[1]s}[%[2]s]))`,
		clusterMatcher,
		nextGenKeyspaceLookupWindow,
	)
}

func ruMatcherForServicePlan(servicePlan string, clusterID uint64, keyspaceID string) (string, error) {
	switch normalizedServicePlan(servicePlan) {
	case strings.ToLower(string(entity.ServicePlanEssential)),
		strings.ToLower(string(entity.ServicePlanStarter)):
		if keyspaceID == "" {
			return "", fmt.Errorf("keyspace id is empty for service plan %s", servicePlan)
		}
		return fmt.Sprintf(`resource_group="%s"`, keyspaceID), nil
	default:
		return tidbClusterRegexMatcherForServicePlan(servicePlan, clusterID), nil
	}
}

func buildServerlessFrontendRUReadPromQL(matcher string) string {
	// Keep this aligned with the Essential/Starter frontend RU chart formula in dbaas-ui.
	return fmt.Sprintf(`(sum(rate(resource_manager_resource_unit_read_request_unit_sum{%s}[1m])) or vector(0))`, matcher)
}

func buildServerlessFrontendRUWritePromQL(matcher string) string {
	// Keep this aligned with the Essential/Starter frontend RU chart formula in dbaas-ui.
	return fmt.Sprintf(`(sum(rate(resource_manager_resource_unit_write_request_unit_sum{%s}[1m])) or vector(0))`, matcher)
}

func buildServerlessFrontendRUTotalPromQL(matcher string) string {
	// Keep this aligned with the Essential/Starter frontend RU chart formula in dbaas-ui.
	return fmt.Sprintf(
		`(sum(rate(resource_manager_resource_unit_read_request_unit_sum{%[1]s}[1m])) or vector(0)) + `+
			`(sum(rate(resource_manager_resource_unit_write_request_unit_sum{%[1]s}[1m])) or vector(0)) + `+
			`(sum(rate(resource_manager_resource_unit_sql_layer_request_unit{%[1]s}[1m])) or vector(0)) + `+
			`(sum(rate(resource_manager_resource_unit_sql_layer_request_unit_sum{%[1]s}[1m])) or vector(0)) + `+
			`(sum(rate(gateway_connection_flow_bytes_sum{namespace="tidb-admin",direction="out",%[1]s}[1m])/(1024)) or vector(0)) + `+
			`(sum(rate(sum(tikv_worker_load_data_wru_cost_counter{%[1]s})[1m])) or vector(0)) + `+
			`(sum(increase(sum(export_ru_cost_counter{%[1]s})[120s])/120) or vector(0))`,
		matcher,
	)
}

func buildRUTotalPromQL(servicePlan string, matcher string) string {
	switch normalizedServicePlan(servicePlan) {
	case strings.ToLower(string(entity.ServicePlanEssential)),
		strings.ToLower(string(entity.ServicePlanStarter)):
		return buildServerlessFrontendRUTotalPromQL(matcher)
	default:
		return fmt.Sprintf(
			`sum by (resource_group) (rate(resource_manager_resource_unit_read_request_unit_sum{%s}[1m])) + sum by (resource_group) (rate(resource_manager_resource_unit_write_request_unit_sum{%s}[1m]))`,
			matcher, matcher,
		)
	}
}

func getNextGenKeyspaceID(ctx context.Context, clusterID uint64, metricName string, queryTime time.Time) (string, error) {
	return getOrResolveNextGenKeyspaceID(clusterID, func() (string, error) {
		clusterMatcher := exactClusterMatcher(ctx, clusterID)
		for _, promql := range []string{
			buildNextGenKeyspaceLookupPromQL(clusterMatcher),
			buildNextGenHistoricalKeyspaceLookupPromQL(clusterMatcher),
		} {
			resp, err := client.DoInstantQueryWithTime(ctx, promql, metricName+".keyspace_lookup", queryTime)
			if err != nil {
				continue
			}
			if resp == nil || resp.Status != "success" || len(resp.Data.Result) == 0 {
				continue
			}

			for _, result := range resp.Data.Result {
				if result.LabelSet.KeyspaceID == "" {
					continue
				}
				return result.LabelSet.KeyspaceID, nil
			}
		}

		return "", fmt.Errorf("failed to resolve keyspace id for cluster %d", clusterID)
	})
}

func getOrResolveNextGenKeyspaceID(clusterID uint64, resolve func() (string, error)) (string, error) {
	if cached, ok := nextGenKeyspaceIDCache.Load(clusterID); ok {
		return cached.(string), nil
	}

	lookup := &nextGenKeyspaceLookup{done: make(chan struct{})}
	actual, loaded := nextGenKeyspaceIDInflight.LoadOrStore(clusterID, lookup)
	if loaded {
		waiting := actual.(*nextGenKeyspaceLookup)
		<-waiting.done
		return waiting.keyspaceID, waiting.err
	}

	lookup.keyspaceID, lookup.err = resolve()
	if lookup.err == nil {
		nextGenKeyspaceIDCache.Store(clusterID, lookup.keyspaceID)
	}
	nextGenKeyspaceIDInflight.Delete(clusterID)
	close(lookup.done)
	return lookup.keyspaceID, lookup.err
}

func sharedSeriesMatcher(ctx context.Context, clusterID uint64, metricName string, queryTime time.Time) (string, error) {
	servicePlan := servicePlanFromContext(ctx)
	switch normalizedServicePlan(string(servicePlan)) {
	case strings.ToLower(string(entity.ServicePlanEssential)),
		strings.ToLower(string(entity.ServicePlanStarter)):
		keyspaceID, err := getNextGenKeyspaceID(ctx, clusterID, metricName, queryTime)
		if err != nil {
			return "", err
		}
		return sharedSeriesMatcherForServicePlan(string(servicePlan), clusterID, keyspaceID)
	default:
		return sharedSeriesMatcherForServicePlan(string(servicePlan), clusterID, "")
	}
}

func useClusterLevelDBLabelsForServicePlan(servicePlan string) bool {
	return entity.ServicePlan(servicePlan).IsInstanceLevel()
}

func dbGroupLabelsForServicePlan(servicePlan string, extra ...string) string {
	if useClusterLevelDBLabelsForServicePlan(servicePlan) {
		return strings.Join(extra, ", ")
	}

	base := []string{"component", "instance"}
	return strings.Join(append(base, extra...), ", ")
}

func dbGroupLabels(ctx context.Context, extra ...string) string {
	return dbGroupLabelsForServicePlan(string(servicePlanFromContext(ctx)), extra...)
}

func sumExprBy(ctx context.Context, expr string, extra ...string) string {
	groupBy := dbGroupLabels(ctx, extra...)
	if groupBy == "" {
		return fmt.Sprintf("sum(%s)", expr)
	}
	return fmt.Sprintf("sum(%s) by (%s)", expr, groupBy)
}

func exactClusterMatcher(ctx context.Context, clusterID uint64) string {
	addr, err := datasource.GetMetricDataAddressFromContext(ctx)
	if err != nil {
		return exactClusterMatcherForServicePlan("", clusterID)
	}

	return exactClusterMatcherForServicePlan(addr.TidbClusterInfo.ServicePlan, clusterID)
}

func aggregateDBMatcher(ctx context.Context, clusterID uint64) string {
	addr, err := datasource.GetMetricDataAddressFromContext(ctx)
	if err != nil {
		return aggregateDBMatcherForServicePlan("", clusterID)
	}

	return aggregateDBMatcherForServicePlan(addr.TidbClusterInfo.ServicePlan, clusterID)
}

func regexClusterMatcher(ctx context.Context, clusterID uint64) string {
	addr, err := datasource.GetMetricDataAddressFromContext(ctx)
	if err != nil {
		return regexClusterMatcherForServicePlan("", clusterID)
	}

	return regexClusterMatcherForServicePlan(addr.TidbClusterInfo.ServicePlan, clusterID)
}

func tidbClusterRegexMatcher(ctx context.Context, clusterID uint64) string {
	addr, err := datasource.GetMetricDataAddressFromContext(ctx)
	if err != nil {
		return tidbClusterRegexMatcherForServicePlan("", clusterID)
	}

	return tidbClusterRegexMatcherForServicePlan(addr.TidbClusterInfo.ServicePlan, clusterID)
}

func QueryNodeMemoryUsedBytes(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	// use node level metrics for memory
	matcher := exactClusterMatcher(ctx, clusterID)
	promql := buildNodeMemoryUsedBytesPromQL(matcher)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func buildNodeMemoryUsedBytesPromQL(matcher string) string {
	return fmt.Sprintf(
		`max by (component, instance) (node_memory_MemTotal_bytes{%[1]s}) - max by (component, instance) (node_memory_MemAvailable_bytes{%[1]s})`,
		matcher,
	)
}

func QueryNodeMemoryTotalBytes(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := fmt.Sprintf("node_memory_MemTotal_bytes{%s}", exactClusterMatcher(ctx, clusterID))
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryNodeMemoryAvailableBytes(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := fmt.Sprintf("node_memory_MemAvailable_bytes{%s}", exactClusterMatcher(ctx, clusterID))
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryNodeCPUUsedSeconds(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	// user node level metrics for cpu
	// promql := fmt.Sprintf("process_cpu_seconds_total{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"} or tiflash_proxy_process_cpu_seconds_total{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID, clusterID)
	promql := fmt.Sprintf("round(sum(node_cpu_seconds_total{%s, mode!= \"idle\"}) by (component, instance))", exactClusterMatcher(ctx, clusterID))
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryNodeCPUUsedSecondsRate(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	// user node level metrics for cpu
	promql := buildNodeCPUUsedSecondsRatePromQL(exactClusterMatcher(ctx, clusterID))
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func buildNodeCPUUsedSecondsRatePromQL(matcher string) string {
	return fmt.Sprintf("sum(rate(node_cpu_seconds_total{mode!=\"idle\", %s}[120s])) by (component, instance)", matcher)
}

func QueryNodeStorageUsedBytes(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	// promql := fmt.Sprintf("(label_replace(tikv_store_size_bytes{instance=~\"db-.*\", kubernetes_namespace=\"tidb%d\",type=\"capacity\"}, \"type\", \"\", \"\", \"\")-label_replace(tikv_store_size_bytes{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\", type=\"available\"}, \"type\", \"\", \"\", \"\")) or (tiflash_system_current_metric_StoreSizeCapacity{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"} - tiflash_system_current_metric_StoreSizeAvailable{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"})",
	// 	clusterID, clusterID, clusterID, clusterID)
	// Dedicated storage usage must stay on the corrected capacity - available path.
	promql := fmt.Sprintf("(tikv_store_size_bytes{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\",type=\"capacity\"} - ignoring(type) tikv_store_size_bytes{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\",type=\"available\"}) or (tiflash_system_current_metric_StoreSizeCapacity{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"} - tiflash_system_current_metric_StoreSizeAvailable{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"})", clusterID, clusterID, clusterID, clusterID)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryNodeStorageCapacityBytes(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	// promql := fmt.Sprintf("sum(pd_cluster_status{type=\"storage_capacity\", kubernetes_namespace=\"tidb%d\"})", clusterID)
	promql := fmt.Sprintf("tikv_store_size_bytes{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\",type=\"capacity\"} or tiflash_system_current_metric_StoreSizeCapacity{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID, clusterID)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryNodeStorageAvailableBytes(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := fmt.Sprintf("tikv_store_size_bytes{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\",type=\"available\"} or tiflash_system_current_metric_StoreSizeAvailable{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID, clusterID)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDBQueriesTotal(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := fmt.Sprintf("tidb_executor_statement_total{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDBFailedQueriesTotal(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := fmt.Sprintf("tidb_server_execute_error_total{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDBConnections(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := fmt.Sprintf("tidb_server_connections{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func queryDBQueryDurationSecondsMetric(ctx context.Context, clusterID uint64, metricName, metricSuffix string, groupBy []string, startTime, endTime time.Time) (*client.Response, error) {
	matcher := aggregateDBMatcher(ctx, clusterID)
	promql := fmt.Sprintf(
		"sum by (%s) (tidb_server_handle_query_duration_seconds_%s{%s, sql_type!=\"internal\"})",
		strings.Join(groupBy, ", "),
		metricSuffix,
		matcher,
	)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDBQueryDurationSecondsSum(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	return queryDBQueryDurationSecondsMetric(ctx, clusterID, metricName, "sum", []string{"sql_type"}, startTime, endTime)
}

func QueryDBQueryDurationSecondsCount(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	return queryDBQueryDurationSecondsMetric(ctx, clusterID, metricName, "count", []string{"sql_type"}, startTime, endTime)
}

func QueryDBQueryDurationSecondsBucket(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	return queryDBQueryDurationSecondsMetric(ctx, clusterID, metricName, "bucket", []string{"sql_type", "le"}, startTime, endTime)
}

func QueryDBQueryDurationSeconds(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	// Keep the historical entrypoint, but use aggregated histogram series to avoid
	// exporting high-cardinality instance-level raw buckets for large Dedicated clusters.
	promql := fmt.Sprintf("sum by (sql_type, le) (tidb_server_handle_query_duration_seconds_bucket{%s, sql_type!=\"internal\"})", aggregateDBMatcher(ctx, clusterID))
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDBTotalConnection(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := sumExprBy(ctx, fmt.Sprintf("tidb_server_connections{%s}", aggregateDBMatcher(ctx, clusterID)))
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDBActiveConnections(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := sumExprBy(ctx, fmt.Sprintf("tidb_server_tokens{%s}", aggregateDBMatcher(ctx, clusterID)))
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDBDisconnections(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := sumExprBy(ctx, fmt.Sprintf("rate(tidb_server_disconnection_total{%s}[2m])", aggregateDBMatcher(ctx, clusterID)), "result")
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDBDatabaseTime(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := sumExprBy(ctx, fmt.Sprintf("rate(tidb_server_handle_query_duration_seconds_sum{%s, sql_type!=\"internal\"}[2m])", aggregateDBMatcher(ctx, clusterID)), "sql_type")
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDBQueryPerSecond(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := sumExprBy(ctx, fmt.Sprintf("rate(tidb_executor_statement_total{%s}[2m])", aggregateDBMatcher(ctx, clusterID)), "type")
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDBFailedQueries(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := sumExprBy(ctx, fmt.Sprintf("rate(tidb_server_execute_error_total{%s}[2m])", aggregateDBMatcher(ctx, clusterID)), "type")
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDBCommandPerSecond(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := sumExprBy(ctx, fmt.Sprintf("rate(tidb_server_query_total{%s}[2m])", aggregateDBMatcher(ctx, clusterID)), "type")
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDBQueriesUsingPlanCacheOps(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := sumExprBy(ctx, fmt.Sprintf("rate(tidb_server_plan_cache_total{%s}[2m])", aggregateDBMatcher(ctx, clusterID)), "type")
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDBAverageQueryDuration(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	matcher := aggregateDBMatcher(ctx, clusterID)
	numerator := sumExprBy(ctx, fmt.Sprintf("rate(tidb_server_handle_query_duration_seconds_sum{%s, sql_type!=\"internal\"}[2m])", matcher), "sql_type")
	denominator := sumExprBy(ctx, fmt.Sprintf("rate(tidb_server_handle_query_duration_seconds_count{%s, sql_type!=\"internal\"}[2m])", matcher), "sql_type")
	promql := fmt.Sprintf("%s / %s > 0", numerator, denominator)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDBTransactionPerSecond(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := sumExprBy(ctx, fmt.Sprintf("rate(tidb_session_transaction_duration_seconds_count{%s, scope=~\"general\"}[2m])", aggregateDBMatcher(ctx, clusterID)), "type", "txn_mode")
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

// cdc related
// Changefeed checkpoint lag
func QueryTiCDCLatencySeconds(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	matcher, err := sharedSeriesMatcher(ctx, clusterID, metricName, startTime)
	if err != nil {
		return nil, err
	}
	promql := fmt.Sprintf("max by(changefeed) (ticdc_owner_checkpoint_ts_lag{%s, changefeed!=\"\"})", matcher)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryTiCDCCheckpointTS(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	matcher, err := sharedSeriesMatcher(ctx, clusterID, metricName, startTime)
	if err != nil {
		return nil, err
	}
	promql := fmt.Sprintf("max by(changefeed) (ticdc_owner_checkpoint_ts{%s, changefeed!=\"\"})", matcher)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

// Changefeed resolved ts lag
func QueryTiCDCResolvedTS(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	matcher, err := sharedSeriesMatcher(ctx, clusterID, metricName, startTime)
	if err != nil {
		return nil, err
	}
	promql := fmt.Sprintf("max by(changefeed) (ticdc_owner_resolved_ts_lag{%s, changefeed!=\"\"})", matcher)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

// Changefeed status
func QueryChangefeedStatus(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	matcher, err := sharedSeriesMatcher(ctx, clusterID, metricName, startTime)
	if err != nil {
		return nil, err
	}
	promql := fmt.Sprintf("max by(changefeed) (ticdc_owner_status{%s, changefeed!=\"\"})", matcher)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryTiCDCReplicaRowsPerSecond(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	matcher, err := sharedSeriesMatcher(ctx, clusterID, metricName, startTime)
	if err != nil {
		return nil, err
	}
	promql := fmt.Sprintf("sum by(changefeed) (rate(ticdc_sink_total_flushed_rows_count{%s,changefeed!=\"\"}[1m])) or sum by(changefeed) (rate(ticdc_sinkv2_batch_row_count_sum{%s, changefeed!=\"\"}[1m])) or sum by(changefeed) (rate(ticdc_sink_batch_row_count_sum{%s,changefeed!=\"\"}[1m]))", matcher, matcher, matcher)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

// DM related
func QueryDMTaskState(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := fmt.Sprintf("dm_worker_task_state{job_id=~\"\"}")
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDMReplicationLagBucket(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := fmt.Sprintf("{__name__=~\"dm_syncer_replication_lag_(bucket|sum|count)\",job_id=~\"\"}")
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDMReplicationLagGauge(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := fmt.Sprintf("dm_syncer_replication_lag_gauge{job_id=~\"\"}")
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryDMRelayReadErrorCount(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := fmt.Sprintf("dm_relay_read_error_count{job_id=~\"\"}")
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryKVRequestDuration(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	matcher := regexClusterMatcher(ctx, clusterID)
	promql := fmt.Sprintf("sum(rate(tidb_tikvclient_request_seconds_sum{store!=\"0\", %s}[2m])) by (instance, type)/ sum(rate(tidb_tikvclient_request_seconds_count{store!=\"0\", %s}[2m])) by (instance,type)", matcher, matcher)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryUptime(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	promql := fmt.Sprintf("(time() - process_start_time_seconds{%s, component=~\"(tidb|tikv)\"}) or tiflash_system_asynchronous_metric_Uptime{cluster_id=~\".*%d\"}", tidbClusterRegexMatcher(ctx, clusterID), clusterID)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

// Control Plane
// metrics to add:
// node_disk_read_time_seconds_total
// node_disk_reads_completed_total
func QueryDiskReadLatency(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	matcher := regexClusterMatcher(ctx, clusterID)
	promql := fmt.Sprintf("rate(node_disk_read_time_seconds_total{%s}[2m]) / rate(node_disk_reads_completed_total{%s}[2m])", matcher, matcher)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

// metrics to add:
// node_disk_write_time_seconds_total
// node_disk_writes_completed_total
func QueryDiskWriteLatency(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	matcher := regexClusterMatcher(ctx, clusterID)
	promql := fmt.Sprintf("rate(node_disk_write_time_seconds_total{%s}[2m]) / rate(node_disk_writes_completed_total{%s}[2m])", matcher, matcher)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

// RU metrics.
func QueryRURead(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	servicePlan := servicePlanFromContext(ctx)

	var promql string
	if normalizedServicePlan(string(servicePlan)) == strings.ToLower(string(entity.ServicePlanEssential)) ||
		normalizedServicePlan(string(servicePlan)) == strings.ToLower(string(entity.ServicePlanStarter)) {
		keyspaceID, err := getNextGenKeyspaceID(ctx, clusterID, metricName, startTime)
		if err != nil {
			return nil, err
		}
		promql = buildServerlessFrontendRUReadPromQL(fmt.Sprintf(`resource_group="%s"`, keyspaceID))
	} else {
		matcher, err := ruMatcherForServicePlan(string(servicePlan), clusterID, "")
		if err != nil {
			return nil, err
		}
		promql = fmt.Sprintf("rate(resource_manager_resource_unit_read_request_unit_sum{%s}[1m])", matcher)
	}
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryRUWrite(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	servicePlan := servicePlanFromContext(ctx)

	var promql string
	if normalizedServicePlan(string(servicePlan)) == strings.ToLower(string(entity.ServicePlanEssential)) ||
		normalizedServicePlan(string(servicePlan)) == strings.ToLower(string(entity.ServicePlanStarter)) {
		keyspaceID, err := getNextGenKeyspaceID(ctx, clusterID, metricName, startTime)
		if err != nil {
			return nil, err
		}
		promql = buildServerlessFrontendRUWritePromQL(fmt.Sprintf(`resource_group="%s"`, keyspaceID))
	} else {
		matcher, err := ruMatcherForServicePlan(string(servicePlan), clusterID, "")
		if err != nil {
			return nil, err
		}
		promql = fmt.Sprintf("rate(resource_manager_resource_unit_write_request_unit_sum{%s}[1m])", matcher)
	}
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}

func QueryRUTotal(ctx context.Context, clusterID uint64, metricName string, startTime, endTime time.Time) (*client.Response, error) {
	servicePlan := servicePlanFromContext(ctx)

	var matcher string
	if normalizedServicePlan(string(servicePlan)) == strings.ToLower(string(entity.ServicePlanEssential)) ||
		normalizedServicePlan(string(servicePlan)) == strings.ToLower(string(entity.ServicePlanStarter)) {
		keyspaceID, err := getNextGenKeyspaceID(ctx, clusterID, metricName, startTime)
		if err != nil {
			return nil, err
		}
		matcher, err = ruMatcherForServicePlan(string(servicePlan), clusterID, keyspaceID)
		if err != nil {
			return nil, err
		}
	} else {
		var err error
		matcher, err = ruMatcherForServicePlan(string(servicePlan), clusterID, "")
		if err != nil {
			return nil, err
		}
	}

	promql := buildRUTotalPromQL(string(servicePlan), matcher)
	if startTime.Equal(endTime) {
		return client.DoInstantQueryWithTime(ctx, promql, metricName, startTime)
	} else {
		return client.DoRangeQuery(ctx, promql, metricName, startTime, endTime)
	}
}