package metadb

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

	"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/metadata"
	"github.com/tidbcloud/metrics-integration/pipeline/internal/otel_collector/extension/taskqueue"
	tidbcomp "github.com/tidbcloud/metrics-integration/pipeline/pkg/tidb/component"
	"go.opentelemetry.io/collector/pdata/pcommon"
	"go.opentelemetry.io/collector/pdata/pmetric"
	"go.opentelemetry.io/collector/receiver"
	"go.uber.org/zap"
)

type metadbDataFetcher struct {
	logger           *zap.Logger
	config           *Config
	integrationTypes []entity.IntegrationType
	mb               *metadata.MetricsBuilder
}

func newDataFetcher(logger *zap.Logger, config *Config, iTypes []entity.IntegrationType, settings receiver.Settings) *metadbDataFetcher {
	return &metadbDataFetcher{
		logger:           logger,
		config:           config,
		integrationTypes: iTypes,
		mb:               metadata.NewMetricsBuilder(config.MetricsBuilderConfig, settings),
	}
}

func (m *metadbDataFetcher) fetchData(ctx context.Context, task taskqueue.Task) (pmetric.Metrics, error) {
	start := task.Timestamp
	end := task.EndTimestamp
	integrationType := ""
	if len(m.integrationTypes) > 0 {
		integrationType = string(m.integrationTypes[0])
	}
	defer func() {
		receiverDuration.
			WithLabelValues(integrationType).Observe(time.Since(start).Seconds())
	}()

	// init resource builder
	rb := m.mb.NewResourceBuilder()
	cs := config.Global().TargetClusters.ToClusterMap()
	nextGenResources := pmetric.NewMetrics()

	validClusterCnt := 0
	// For each cluster, build a standalone ResourceMetric.
	for id, c := range cs {

		clusterIdString := strconv.FormatUint(id, 10)
		rb.SetClusterID(clusterIdString)
		rb.SetClusterName(c.Name)
		projectIdString := strconv.FormatUint(c.ProjectId, 10)
		rb.SetProjectID(projectIdString)
		projectName := config.Global().TargetClusters.GetProjectName(c.ProjectId)
		rb.SetProjectName(projectName)
		res := rb.Emit()

		compMap := c.NodeCPUCapacityComponentMap()

		tidbClusterInfo := datasource.TidbClusterInfo{
			Vendor:      c.CloudProviderName,
			Region:      c.RegionName,
			TenantID:    entity.ID(c.TenantId),
			ProjectID:   entity.ID(c.ProjectId),
			ClusterID:   entity.ID(c.Id),
			ServicePlan: c.ServicePlan,
		}
		ctx, err := datasource.SetMetricDataAddressToContextWithTidbClusterInfo(ctx, tidbClusterInfo)
		if err != nil {
			m.logger.Error("failed to call datasource.SetMetricDataAddressToContextWithTidbClusterInfo, skip this tidb cluster",
				zap.String("tidb_cluster_id", clusterIdString), zap.Error(err))
			continue
		} else {
			addr, err := datasource.GetMetricDataAddressFromContext(ctx)
			m.logger.Debug("get metrics data address for cluster",
				zap.String("tidb_cluster_id", clusterIdString),
				zap.String("metrics url", addr.QueryUrl),
				zap.Error(err),
			)
		}

		// NextGen clusters (Premium/Essential/Starter) don't have node-level metrics.
		// Skip QueryNodeMemoryUsedBytes-based instance enumeration for them, but
		// still emit the ResourceMetric so the downstream thanos processor sees
		// the cluster and can query shared DB/RU metrics plus any plan-supported CDC metrics.
		sp := entity.ServicePlan(c.ServicePlan)
		if sp.IsNextGen() {
			validClusterCnt++
			rm := nextGenResources.ResourceMetrics().AppendEmpty()
			res.CopyTo(rm.Resource())
			continue
		}

		// Get cluster's instances.
		// TODO: make tidb_cloud.node_memory_used_bytes encapsuled in some place.
		thanosRes, err := central_thanos.QueryNodeMemoryUsedBytes(ctx, c.Id, "tidb_cloud.node_memory_used_bytes", end, end)
		if err != nil || thanosRes.Status != "success" {
			err = fmt.Errorf("err: %v, result: %+v", err, thanosRes)
			m.logger.Error("failed to query used node memory, skip this tidb cluster",
				zap.String("tidb_cluster_id", clusterIdString), zap.Error(err))
			continue
		} else {
			m.logger.Debug("thanos res fetched for cluster",
				zap.String("tidb_cluster_id", clusterIdString),
				zap.Int("result len", len(thanosRes.Data.Result)),
			)
			validClusterCnt++
		}

		// For each instance, create memory/cpu/disk capacity metrics.
		for _, result := range thanosRes.Data.Result {
			labelSet := result.LabelSet
			valueTuples := result.Values
			for _, valueTuple := range valueTuples {
				if len(valueTuple) != 2 {
					err := fmt.Errorf("failed to query value(%+v) of used node memory", valueTuple)
					m.logger.Error("failed to query used node memory, skip this tidb cluster",
						zap.String("tidb_cluster_id", clusterIdString), zap.Error(err))
					break
				}
				if labelSet.Component == tidbcomp.PD.String() {
					continue
				}

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

				capacityValue, ok := compMap[labelSet.Component]
				if !ok {
					continue
				}
				m.mb.RecordTidbCloudNodeCPUCapacityCoresDataPoint(pcommon.NewTimestampFromTime(preciseTime), capacityValue, c.Name, clusterIdString, datasource.UnifyDBName(labelSet.Instance), metadata.MapAttributeComponent[labelSet.Component], projectName)
			}
		}

		m.mb.EmitForResource(metadata.WithResource(res))
	}

	result := m.mb.Emit()
	for i := 0; i < nextGenResources.ResourceMetrics().Len(); i++ {
		nextGenResources.ResourceMetrics().At(i).MoveTo(result.ResourceMetrics().AppendEmpty())
	}

	if result.ResourceMetrics().Len() == 0 {
		return result, fmt.Errorf("no valid clusters found")
	}

	return result, nil
}