package thanos

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/metrics-integration/pipeline/central/config"
	"github.com/tidbcloud/metrics-integration/pipeline/central/entity"
	"github.com/tidbcloud/metrics-integration/pipeline/internal/datasource"
	"github.com/tidbcloud/metrics-integration/pipeline/internal/datasource/central_thanos"
	"github.com/tidbcloud/metrics-integration/pipeline/internal/datasource/client"
	"github.com/tidbcloud/metrics-integration/pipeline/internal/metadata"
	"github.com/tidbcloud/metrics-integration/pipeline/internal/otel_collector"
	"github.com/tidbcloud/metrics-integration/pipeline/internal/otel_collector/extension/taskqueue"
	"go.opentelemetry.io/collector/component"
	"go.opentelemetry.io/collector/consumer"
	"go.opentelemetry.io/collector/pdata/pcommon"
	"go.opentelemetry.io/collector/pdata/pmetric"
	"go.opentelemetry.io/collector/processor"
	"go.opentelemetry.io/collector/receiver"
	"go.uber.org/zap"
)

var (
	consumerCapabilities = consumer.Capabilities{MutatesData: true}
	// Same as the metric in tidb: https://github.com/pingcap/tidb/blob/master/metrics/server.go#L45.
	durationBuckets = prometheus.ExponentialBuckets(0.0005, 2, 29)

	processingDuration = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "otelcol_processor_query_duration",
			Help:    "the time that used for querying metrics from o11y metrics backend",
			Buckets: prometheus.ExponentialBuckets(0.5, 2, 8), // 500ms ~ 128s
		}, []string{"integration_type"})
)

func init() {
	prometheus.MustRegister(processingDuration)
}

type centralThanosQuerier struct {
	config *Config

	logger *zap.Logger

	taskQueue taskqueue.Consumer

	mb *metadata.MetricsBuilder

	setMetricDataAddressToContext func(ctx context.Context, tidbClusterID entity.ID) (context.Context, error)
	queryThenRecordFunc           func(
		ctx context.Context,
		start time.Time,
		end time.Time,
		metricName string,
		clusterInfo infoTp,
		queryFunc func(ctx context.Context, clusterID uint64, metric string, start, end time.Time) (*client.Response, error),
		recordFunc func(ts pcommon.Timestamp, value any, labels client.LabelSet, info infoTp) error,
	) (lastExistTimestamp time.Time, err error)
}

func newCentralThanosQuerier(logger *zap.Logger, config *Config, settings processor.Settings) (*centralThanosQuerier, error) {
	mbConfig := config.MetricsBuilderConfig

	// overide resource attribute settings.
	mbConfig.ResourceAttributes = metadata.ResourceAttributesConfig{
		ClusterID: metadata.ResourceAttributeConfig{
			Enabled: true,
		},
		ClusterName: metadata.ResourceAttributeConfig{
			Enabled: true,
		},
		ProjectID: metadata.ResourceAttributeConfig{
			Enabled: true,
		},
		ProjectName: metadata.ResourceAttributeConfig{
			Enabled: true,
		},
	}
	querier := centralThanosQuerier{
		config: config,
		logger: logger,
		mb:     metadata.NewMetricsBuilder(mbConfig, receiver.Settings(settings)),
	}
	querier.setMetricDataAddressToContext = datasource.SetMetricDataAddressToContextWithCache
	querier.queryThenRecordFunc = querier.queryThenRecord
	return &querier, nil
}

func (q *centralThanosQuerier) start(ctx context.Context, host component.Host) error {
	q.logger.Info("get extension id", zap.Any("extension_id", q.config.TaskQueue))
	taskQueueExt, ok := host.GetExtensions()[q.config.TaskQueue]
	if !ok {
		q.logger.Info("taskqueue not found, backfill mode is not enabled")
	} else {
		q.taskQueue = taskQueueExt.(taskqueue.Consumer)
	}

	return nil

}

func (q *centralThanosQuerier) nackTask(task *taskqueue.Task) {
	if task == nil || q.taskQueue == nil {
		return
	}
	if err := taskqueue.NackWithRequeue(q.taskQueue, *task); err != nil {
		q.logger.Warn("failed to nack task", zap.Error(err))
	}
}

// This function assumes the upstream receiver can only be metadb receiver.
func (q *centralThanosQuerier) doQuery(ctx context.Context, md pmetric.Metrics) (pmetric.Metrics, error) {
	start := time.Now()
	end := start
	task, taskExist := taskqueue.GetTaskFromContext(ctx)
	if taskExist {
		start = task.Timestamp
		end = task.EndTimestamp
	}
	// integrationType will be updated during
	integrationType := ""
	// we consider integrationType for all resourceMetrics are the same in a single doQuery
	if md.ResourceMetrics().Len() > 0 {
		if _integrationType, ok := md.ResourceMetrics().At(0).Resource().Attributes().Get(otel_collector.IntegrationType); ok {
			integrationType = _integrationType.AsString()
		}
	}
	defer func() {
		processingDuration.
			WithLabelValues(integrationType).Observe(time.Since(start).Seconds())
	}()

	lastExistTimestamp := time.Time{}
	clusterMap := config.Global().TargetClusters.ToClusterMap()
	servicePlanByClusterID := make(map[string]entity.ServicePlan, md.ResourceMetrics().Len())

	for i := 0; i < md.ResourceMetrics().Len(); i++ {
		rm := md.ResourceMetrics().At(i)
		// now := pcommon.NewTimestampFromTime(time.Now())

		// Access meta attributes generated by upstream metadb receiver.
		resources := rm.Resource()
		attrs := resources.Attributes()
		_id, ok := attrs.Get("cluster_id")
		if !ok {
			// Log then go to the next cluster id.
			q.logger.Warn("failed to get cluster id")
			continue
		}
		id := _id.AsString()
		_name, ok := attrs.Get("cluster_name")
		if !ok {
			q.logger.Warn("failed to get cluster name")
			continue
		}
		name := _name.AsString()
		clusterID, err := strconv.ParseUint(id, 10, 64)
		if err != nil {
			// Log then go to the next cluster id.
			q.logger.Warn("Failed to parse cluster id", zap.Error(err))
			continue
		}
		projname := ""
		if _projname, ok := attrs.Get("project_name"); ok {
			projname = _projname.AsString()
		}

		// Read the service plan passed by the metadb receiver.
		servicePlan := ""
		if _sp, ok := attrs.Get("service_plan"); ok {
			servicePlan = _sp.AsString()
		}
		sp := entity.ServicePlan(servicePlan)
		if c, ok := clusterMap[clusterID]; ok {
			if sp == "" {
				sp = entity.ServicePlan(c.ServicePlan)
			}
			if projname == "" {
				projname = c.ProjectName
			}
		}
		servicePlanByClusterID[id] = sp

		ctx, err := q.setMetricDataAddressToContext(ctx, entity.ID(clusterID))
		if err != nil {
			// Log then go to the next cluster id.
			q.logger.Warn("Failed to call datasource.SetMetricDataAddressToContextWithCache", zap.Error(err))
			continue
		}

		info := infoTp{
			clusterId:   clusterID,
			clusterName: name,
			projectName: projname,
			servicePlan: string(sp),
		}

		// Query TiDB cluster metrics one by one. Will change to parallel in the future.
		//
		// Metrics are grouped by service plan:
		//   - Shared (all plans): DB performance metrics + RU metrics
		//   - CDC-enabled plans:  CDC metrics
		//   - Dedicated-only:     node memory/CPU/storage, component uptime, DM, KV request, disk latency
		//   - NextGen-only:       row/columnar storage

		// ---- Dedicated-only: node-level, infra, DM metrics ----
		if !sp.IsNextGen() {

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.node_memory_capacity_byte", info, central_thanos.QueryNodeMemoryTotalBytes, q.RecordTidbCloudNodeMemoryCapacityBytes); localErr != nil {
				q.logger.Warn("QueryNodeMemoryCapacityBytes failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.node_memory_used_bytes", info, central_thanos.QueryNodeMemoryUsedBytes, q.RecordTidbCloudNodeMemoryUsedBytes); localErr != nil {
				q.logger.Warn("QueryNodeMemoryUsedBytes failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.node_memory_available_bytes", info, central_thanos.QueryNodeMemoryAvailableBytes, q.RecordTidbCloudNodeMemoryAvailableBytes); localErr != nil {
				q.logger.Warn("QueryNodeMemoryAvailableBytes failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			// get node_cpu_seconds_total for datadog
			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.node_cpu_seconds_total", info, central_thanos.QueryNodeCPUUsedSeconds, q.RecordQueryNodeCPUUsedSeconds); localErr != nil {
				q.logger.Warn("QueryNodeCPUUsedSeconds failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			// get node_cpu_seconds_total for new relic
			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.node_cpu_seconds_total_rate", info, central_thanos.QueryNodeCPUUsedSecondsRate, q.RecordQueryNodeCPUUsedSecondsRate); localErr != nil {
				q.logger.Warn("QueryNodeCPUUsedSecondsRate failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.node_storage_capacity_bytes", info, central_thanos.QueryNodeStorageCapacityBytes, q.RecordQueryNodeStorageCapacityBytes); localErr != nil {
				q.logger.Warn("QueryNodeStorageCapacityBytes failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.node_storage_used_bytes", info, central_thanos.QueryNodeStorageUsedBytes, q.RecordQueryNodeStorageUsedBytes); localErr != nil {
				q.logger.Warn("QueryNodeStorageUsedBytes failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.node_storage_available_bytes", info, central_thanos.QueryNodeStorageAvailableBytes, q.RecordQueryNodeStorageAvailableBytes); localErr != nil {
				q.logger.Warn("QueryNodeStorageAvailableBytes failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.tidb_cloud.db_component_uptime", info, central_thanos.QueryUptime, q.RecordQueryDBComponentUptime); localErr != nil {
				q.logger.Warn("QueryDBCompoenentUptime failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.dm_task_state", info, central_thanos.QueryDMTaskState, q.RecordQueryDmTaskState); localErr != nil {
				q.logger.Warn("QueryDMTaskStatus failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			if _, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.dm_syncer_replication_lag_bucket", info, central_thanos.QueryDMReplicationLagBucket, q.RecordDMReplicationLagBucket); localErr != nil {
				q.logger.Warn("QueryDMReplicationLagBucket failed", zap.Error(localErr))
			}

			if _, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.dm_syncer_replication_lag_gauge", info, central_thanos.QueryDMReplicationLagGauge, q.RecordDMReplicationLagGauge); localErr != nil {
				q.logger.Warn("QueryDMReplicationLagGauge failed", zap.Error(localErr))
			}

			if _, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.dm_relay_read_error_count", info, central_thanos.QueryDMRelayReadErrorCount, q.RecordDMRelayReadErrorCount); localErr != nil {
				q.logger.Warn("QueryDMRelayReadErrorCount failed", zap.Error(localErr))
			}

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.db_kv_request_duration", info, central_thanos.QueryKVRequestDuration, q.RecordDBQueryKVRequestDuration); localErr != nil {
				q.logger.Warn("KVRequestDuration failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
				}
				return md, localErr
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.node_disk_read_latency", info, central_thanos.QueryDiskReadLatency, q.RecordTidbCloudNodeDiskReadLatency); localErr != nil {
				q.logger.Warn("NodeDiskReadLatency failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.node_disk_write_latency", info, central_thanos.QueryDiskWriteLatency, q.RecordTidbCloudNodeDiskWriteLatency); localErr != nil {
				q.logger.Warn("NodeDiskWriteLatency failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

		} // end Dedicated-only

		// ---- Shared (all plans): DB performance metrics ----

		if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.db_total_connection", info, central_thanos.QueryDBTotalConnection, q.RecordQueryDBTotalConnection); localErr != nil {
			q.logger.Warn("QueryDBTotalConnection failed", zap.Error(localErr))
			if taskExist {
				q.nackTask(task)
				return md, localErr
			}
		} else {
			if metricLastTs.After(lastExistTimestamp) {
				lastExistTimestamp = metricLastTs
			}
		}

		if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.db_active_connections", info, central_thanos.QueryDBActiveConnections, q.RecordQueryDBActiveConnections); localErr != nil {
			q.logger.Warn("QueryDBActiveConnections failed", zap.Error(localErr))
			if taskExist {
				q.nackTask(task)
				return md, localErr
			}
		} else {
			if metricLastTs.After(lastExistTimestamp) {
				lastExistTimestamp = metricLastTs
			}
		}

		if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.db_disconnections", info, central_thanos.QueryDBDisconnections, q.RecordQueryDBDisconnections); localErr != nil {
			q.logger.Warn("QueryDBDisconnections failed", zap.Error(localErr))
			if taskExist {
				q.nackTask(task)
				return md, localErr
			}
		} else {
			if metricLastTs.After(lastExistTimestamp) {
				lastExistTimestamp = metricLastTs
			}
		}

		if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.db_database_time", info, central_thanos.QueryDBDatabaseTime, q.RecordQueryDBDatabaseTime); localErr != nil {
			q.logger.Warn("QueryDBDatabaseTime failed", zap.Error(localErr))
			if taskExist {
				q.nackTask(task)
				return md, localErr
			}
		} else {
			if metricLastTs.After(lastExistTimestamp) {
				lastExistTimestamp = metricLastTs
			}
		}

		if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.db_query_per_second", info, central_thanos.QueryDBQueryPerSecond, q.RecordQueryDBQueryPerSecond); localErr != nil {
			q.logger.Warn("QueryDBQueryPerSecond failed", zap.Error(localErr))
			if taskExist {
				q.nackTask(task)
				return md, localErr
			}
		} else {
			if metricLastTs.After(lastExistTimestamp) {
				lastExistTimestamp = metricLastTs
			}
		}

		if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.db_failed_queries", info, central_thanos.QueryDBFailedQueries, q.RecordQueryDBFailedQueries); localErr != nil {
			q.logger.Warn("QueryDBFailedQueries failed", zap.Error(localErr))
			if taskExist {
				q.nackTask(task)
				return md, localErr
			}
		} else {
			if metricLastTs.After(lastExistTimestamp) {
				lastExistTimestamp = metricLastTs
			}
		}

		if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.db_command_per_second", info, central_thanos.QueryDBCommandPerSecond, q.RecordQueryDBCommandPerSecond); localErr != nil {
			q.logger.Warn("QueryDBCommandPerSecond failed", zap.Error(localErr))
			if taskExist {
				q.nackTask(task)
				return md, localErr
			}
		} else {
			if metricLastTs.After(lastExistTimestamp) {
				lastExistTimestamp = metricLastTs
			}
		}

		if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.db_queries_using_plan_cache_ops", info, central_thanos.QueryDBQueriesUsingPlanCacheOps, q.RecordQueryDBQueriesUsingPlanCacheOps); localErr != nil {
			q.logger.Warn("QueryDBQueriesUsingPlanCacheOps failed", zap.Error(localErr))
			if taskExist {
				q.nackTask(task)
				return md, localErr
			}
		} else {
			if metricLastTs.After(lastExistTimestamp) {
				lastExistTimestamp = metricLastTs
			}
		}

		if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.db_average_query_duration", info, central_thanos.QueryDBAverageQueryDuration, q.RecordQueryDBAverageQueryDuration); localErr != nil {
			q.logger.Warn("QueryDBAverageQueryDuration failed", zap.Error(localErr))
			if taskExist {
				q.nackTask(task)
				return md, localErr
			}
		} else {
			if metricLastTs.After(lastExistTimestamp) {
				lastExistTimestamp = metricLastTs
			}
		}

		if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.db_transaction_per_second", info, central_thanos.QueryDBTransactionPerSecond, q.RecordQueryDBTransactionPerSecond); localErr != nil {
			q.logger.Warn("QueryDBTransactionPerSecond failed", zap.Error(localErr))
			if taskExist {
				q.nackTask(task)
				return md, localErr
			}
		} else {
			if metricLastTs.After(lastExistTimestamp) {
				lastExistTimestamp = metricLastTs
			}
		}

		// ---- Shared (all plans): RU metrics ----
		if sp.IsNextGen() {
			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.resource_manager_resource_request_unit_total", info, central_thanos.QueryRUTotal, q.RecordQueryResourceRequestUnitTotal); localErr != nil {
				q.logger.Warn("QueryRUTotal failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}
		} else {
			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.resource_manager_resource_unit_read_request_unit", info, central_thanos.QueryRURead, q.RecordQueryResourceUnitReadRequestUnit); localErr != nil {
				q.logger.Warn("QueryRURead failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.resource_manager_resource_unit_write_request_unit", info, central_thanos.QueryRUWrite, q.RecordQueryResourceUnitWriteRequestUnit); localErr != nil {
				q.logger.Warn("QueryRUWrite failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}
		}

		// ---- CDC-enabled plans only ----
		if sp.SupportsCDC() {

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.cdc_changefeed_latency", info, central_thanos.QueryTiCDCLatencySeconds, q.RecordQueryCdcChangefeedLatency); localErr != nil {
				q.logger.Warn("QueryCdcChangefeedLatency failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			if !sp.IsNextGen() {
				if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.cdc_changefeed_resolved_ts_lag", info, central_thanos.QueryTiCDCResolvedTS, q.RecordQueryCdcChangefeedResolvedTsLag); localErr != nil {
					q.logger.Warn("QueryCdcChangefeedResolvedTsLag failed", zap.Error(localErr))
					if taskExist {
						q.nackTask(task)
						return md, localErr
					}
				} else {
					if metricLastTs.After(lastExistTimestamp) {
						lastExistTimestamp = metricLastTs
					}
				}
			}

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.cdc_changefeed_status", info, central_thanos.QueryChangefeedStatus, q.RecordQueryCdcChangefeedStatus); localErr != nil {
				q.logger.Warn("QueryCdcChangefeedStatus failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			if !sp.IsNextGen() {
				if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.cdc_changefeed_replica_rows", info, central_thanos.QueryTiCDCReplicaRowsPerSecond, q.RecordQueryCdcChangefeedRelicaRows); localErr != nil {
					q.logger.Warn("QueryCdcChangefeedRelicaRows failed", zap.Error(localErr))
					if taskExist {
						q.nackTask(task)
						return md, localErr
					}
				} else {
					if metricLastTs.After(lastExistTimestamp) {
						lastExistTimestamp = metricLastTs
					}
				}
			}
		}

		// ---- NextGen-only: storage metrics ----
		if sp.IsNextGen() {

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.db_row_storage_used_bytes", info, central_thanos.QueryDBRowStorageUsedBytes, q.RecordTidbCloudDbRowStorageUsedBytes); localErr != nil {
				q.logger.Warn("QueryDBRowStorageUsedBytes failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

			if metricLastTs, localErr := q.queryThenRecordFunc(ctx, start, end, "tidb_cloud.db_columnar_storage_used_bytes", info, central_thanos.QueryDBColumnarStorageUsedBytes, q.RecordTidbCloudDbColumnarStorageUsedBytes); localErr != nil {
				q.logger.Warn("QueryDBColumnarStorageUsedBytes failed", zap.Error(localErr))
				if taskExist {
					q.nackTask(task)
					return md, localErr
				}
			} else {
				if metricLastTs.After(lastExistTimestamp) {
					lastExistTimestamp = metricLastTs
				}
			}

		} // end NextGen-only

		q.mb.EmitForResource(metadata.WithResource(resources))
	}

	newMd := q.mb.Emit()

	if taskExist {
		if lastExistTimestamp.IsZero() {
			q.logger.Warn("no datapoint exists, should requeue and wait.",
				zap.Time("task_timestamp", task.Timestamp),
				zap.Time("task_end_timestamp", task.EndTimestamp),
				zap.Int("retried_count", task.GetRetriedCount()),
			)
			q.nackTask(task)
			return md, fmt.Errorf("no datapoint queried")
		} else {
			task.AckTimestamp = lastExistTimestamp.Truncate(task.Step)
			taskqueue.InjectTaskIntoCtx(ctx, task)
		}
	}

	result := mergeMetrics(newMd, md)

	// Apply service-plan-specific attribute transformations for NextGen plans.
	// Must run after mergeMetrics because merge joins on the "cluster_id" resource attribute.
	transformMetricsForServicePlan(result, servicePlanByClusterID)
	stripInternalResourceAttrs(result)

	return result, nil
}

func (q *centralThanosQuerier) queryThenRecord(
	ctx context.Context,
	start time.Time,
	end time.Time,
	metricName string,
	clusterInfo infoTp,
	queryFunc func(ctx context.Context, clusterID uint64, metric string, start, end time.Time) (*client.Response, error),
	recordFunc func(ts pcommon.Timestamp, value any, labels client.LabelSet, info infoTp) error,
) (lastExistTimestamp time.Time, err error) {
	res, err := queryFunc(ctx, clusterInfo.clusterId, metricName, start, end)
	if err != nil || res.Status != "success" {
		return time.Time{}, fmt.Errorf("failed to call queryFunc, err: %v, result: %+v", err, res)
	}

	for _, series := range res.Data.Result {
		labels, values := series.LabelSet, series.Values
		for _, value := range values {
			if len(value) != 2 {
				return time.Time{}, fmt.Errorf("prometheus response failed: value(%+v)", value)
			}

			tsFloat, _ := value[0].(float64)
			nanos := int64(tsFloat-float64(int64(tsFloat))) * 1e9
			preciseTime := time.Unix(int64(tsFloat), nanos)

			err = recordFunc(pcommon.NewTimestampFromTime(preciseTime), value[1], labels, clusterInfo)

			if err != nil {
				return time.Time{}, fmt.Errorf("failed when recording: %w, result: %+v", err, res)
			}
			if preciseTime.After(lastExistTimestamp) {
				lastExistTimestamp = preciseTime
			}
		}

	}

	return lastExistTimestamp, nil
}

func (q *centralThanosQuerier) queryThenRecordHistogram(
	ctx context.Context,
	t time.Time,
	metricName string,
	clusterInfo infoTp,
	queryFunc func(ctx context.Context, clusterID uint64, metric string, start, end time.Time) (*client.Response, error),
	recordFunc func(ts pcommon.Timestamp, histogramData any, labels client.LabelSet, info infoTp) error,
) error {
	res, err := queryFunc(ctx, clusterInfo.clusterId, metricName, t, t)
	if err != nil || res.Status != "success" {
		return fmt.Errorf("failed to call queryFunc, err: %v, result: %+v", err, res)
	}
	histoMap := q.calculateBuckets(res.Data.Result)

	for desc, data := range histoMap {

		if data.t.IsZero() {
			continue
		}

		err = recordFunc(pcommon.NewTimestampFromTime(data.t), data, desc, clusterInfo)
		if err != nil {
			return err
		}
	}

	return nil
}