package metadb

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

	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/metrics-server/central/db/entity"
	"github.com/tidbcloud/metrics-server/central/model/integration"
	centralutil "github.com/tidbcloud/metrics-server/central/model/util"
	"github.com/tidbcloud/metrics-server/central/monitor_data_migration"
	"github.com/tidbcloud/metrics-server/idl/pbgen/proto/core"
	"github.com/tidbcloud/metrics-server/internal/datasource"
	"github.com/tidbcloud/metrics-server/internal/datasource/central_thanos"
	"github.com/tidbcloud/metrics-server/internal/metadata"
	"github.com/tidbcloud/metrics-server/internal/otel_collector"
	tidbcomp "github.com/tidbcloud/metrics-server/pkg/tidb/component"
	"go.opentelemetry.io/collector/component"
	"go.opentelemetry.io/collector/config"
	"go.opentelemetry.io/collector/consumer"
	"go.opentelemetry.io/collector/pdata/pcommon"
	"go.opentelemetry.io/collector/pdata/pmetric"
	"go.opentelemetry.io/collector/receiver/scraperhelper"
	"go.uber.org/zap"
)

const (
	typeStr = "metadb"
)

var (
	receiverDuration = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "otelcol_receiver_scrape_duration",
			Help:    "the time that used for scrape integration info from rds",
			Buckets: prometheus.ExponentialBuckets(0.5, 2, 10), // 500ms ~ 512s
		}, []string{"integration_type"})
)

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

type Config struct {
	scraperhelper.ScraperControllerSettings `mapstructure:",squash"`
	Integration                             Integration `mapstructure:"integration"`
}

func (cfg *Config) Validate() error {
	if err := cfg.ScraperControllerSettings.Validate(); err != nil {
		return err
	}
	return cfg.Integration.Validate()

}

type Integration struct {
	ProviderTypes []string `mapstructure:"provider_type"`
}

func (i *Integration) Validate() error {
	if len(i.ProviderTypes) == 0 {
		return fmt.Errorf("provider_types is empty")
	}
	if len(i.ProviderTypes) != 1 {
		return fmt.Errorf("provider_types only support one provider")
	}
	return nil
}

func NewFactory() component.ReceiverFactory {
	return component.NewReceiverFactory(
		typeStr,
		createDefaultConfig,
		component.WithMetricsReceiver(createMetricsReceiver, component.StabilityLevelStable))
}

func createDefaultConfig() config.Receiver {
	return &Config{
		ScraperControllerSettings: scraperhelper.ScraperControllerSettings{
			ReceiverSettings: config.NewReceiverSettings(config.NewComponentID(typeStr)),
			// https://github.com/pingcap/tidb-operator/blob/master/charts/tidb-cluster/templates/config/_prometheus-config.tpl
			CollectionInterval: 15 * time.Second,
		},
		Integration: Integration{
			ProviderTypes: []string{"DATADOG"},
		},
	}
}

func createMetricsReceiver(
	_ context.Context,
	params component.ReceiverCreateSettings,
	rConf config.Receiver,
	consumer consumer.Metrics,
) (component.MetricsReceiver, error) {
	cfg := rConf.(*Config)

	// We only support Datadog and newrelic integration for now.
	iTypes := make([]entity.IntegrationType, 0)
	for _, providerType := range cfg.Integration.ProviderTypes {
		if providerType == "DATADOG" || providerType == "NEWRELIC" {
			iTypes = append(iTypes, entity.IntegrationType(providerType))
		} else {
			return nil, fmt.Errorf("unsupported integration type: %s", providerType)
		}
	}

	ns := newMetadbScraper(params.Logger, cfg, iTypes)
	scraper, err := scraperhelper.NewScraper(typeStr, ns.scrape)
	if err != nil {
		return nil, err
	}

	return scraperhelper.NewScraperControllerReceiver(
		&cfg.ScraperControllerSettings, params, consumer,
		scraperhelper.AddScraper(scraper),
	)
}

type metadbScraper struct {
	logger           *zap.Logger
	config           *Config
	integrationTypes []entity.IntegrationType
}

func newMetadbScraper(logger *zap.Logger, config *Config, iTypes []entity.IntegrationType) *metadbScraper {
	return &metadbScraper{
		logger:           logger,
		config:           config,
		integrationTypes: iTypes,
	}
}

func (m *metadbScraper) scrape(ctx context.Context) (pmetric.Metrics, error) {
	start := time.Now()
	integrationType := ""
	if len(m.integrationTypes) > 0 {
		integrationType = string(m.integrationTypes[0])
	}
	defer func() {
		receiverDuration.
			WithLabelValues(integrationType).Observe(time.Since(start).Seconds())
	}()
	// Get ids of projects whose integration type is matched and status is active.
	integrations, err := getIntegrations(m.integrationTypes)
	if err != nil {
		return pmetric.Metrics{}, err
	}
	if len(integrations) < 1 {
		return pmetric.Metrics{}, fmt.Errorf("no integration for %v found, thus no metrics will be scraped", m.integrationTypes)
	}
	integrationMap, prjIDs := parseIntegrations(integrations)
	// Get cluster by project ids.
	cs, err := getClusters(unpackEntityIDs(prjIDs))
	m.logger.Info("after call getClusters",
		zap.Int("integrations_count", len(integrations)),
		zap.Any("project_ids", prjIDs),
		zap.Int("clusters_count", len(cs)),
		zap.Error(err))
	if err != nil {
		return pmetric.Metrics{}, err
	}

	projectMigrationStatuses := monitor_data_migration.GetDatadogIntegrationProjectMigrationStatus(prjIDs)

	md := pmetric.NewMetrics()

	// For each cluster, build a standalone ResourceMetric.
	for id, c := range cs {
		m.logger.Info("build ResourceMetric for cluster", zap.Uint64("cluster_id", id))
		// Otel metric initialization.
		rs := md.ResourceMetrics().AppendEmpty()
		ilm := rs.ScopeMetrics().AppendEmpty()
		ilm.Scope().SetName("otel/tidbcloud/metadb")
		now := pcommon.NewTimestampFromTime(time.Now())

		memSlice := ilm.Metrics().AppendEmpty()
		metadata.M.TidbCloudNodeMemoryCapacityBytes.Init(memSlice)
		cpuSlice := ilm.Metrics().AppendEmpty()
		metadata.M.TidbCloudNodeCPUCapacityCores.Init(cpuSlice)
		diskSlice := ilm.Metrics().AppendEmpty()
		metadata.M.TidbCloudNodeStorageCapacityBytes.Init(diskSlice)
		id := strconv.FormatUint(id, 10)
		prjID := strconv.FormatUint(c.ProjectId, 10)

		projectName, err := getProjectName(ctx, c.ProjectId)
		if err != nil {
			m.logger.Error("failed to call getProjectName, skip this tidb cluster",
				zap.String("tidb_cluster_id", id), zap.Error(err))
			continue
		} else {
			m.logger.Info("getProjectName successfully",
				zap.Uint64("project_id", c.ProjectId),
				zap.String("project_name", projectName))
		}

		// Save cluster ids and datadog or newrelic api keys in the resource attribute map.
		// These attributes are used by downstream pipeline.
		attrs := rs.Resource().Attributes()
		attrs.InsertString(otel_collector.ProjectIDKey, prjID)
		attrs.InsertString(otel_collector.ProjectNameKey, projectName)
		attrs.InsertString(otel_collector.ClusterIDKey, id)
		attrs.InsertString(otel_collector.ClusterNameKey, c.Name)

		if len(integrationMap[entity.ID(c.ProjectId)].EncryptedDatadogApiKey) > 0 {
			attrs.InsertString(otel_collector.ApiKeyKey, integrationMap[entity.ID(c.ProjectId)].EncryptedDatadogApiKey)
			attrs.InsertString(otel_collector.SiteKey, integration.GetDatadogSiteUrl(integration.DatadogSiteDBToRest(integrationMap[entity.ID(c.ProjectId)].DatadogSite)))
		} else if len(integrationMap[entity.ID(c.ProjectId)].EncryptedNewrelicApiKey) > 0 {
			attrs.InsertString(otel_collector.ApiKeyKey, integrationMap[entity.ID(c.ProjectId)].EncryptedNewrelicApiKey)
			attrs.InsertString(otel_collector.SiteKey, integration.GetNewrelicSiteEndpoint(integration.NewrelicSiteDBToRest(integrationMap[entity.ID(c.ProjectId)].NewrelicSite)))
		}

		compMap := make(map[string]struct {
			mem  int64
			cpu  float64
			disk int64
		})
		for _, comp := range c.Components {
			if comp.ComponentType == core.ComponentType_COMPONENT_TYPE_PD {
				continue
			}
			name := centralutil.CoreComponentType2componentName(comp.ComponentType).String()
			compMap[name] = struct {
				mem  int64
				cpu  float64
				disk int64
			}{
				mem:  int64(comp.MemoryGi) * 1024 * 1024 * 1024,
				cpu:  float64(comp.CpuNum),
				disk: int64(comp.StorageSizeGi) * 1024 * 1024 * 1024,
			}
		}

		// Get metric data source address
		projectMigrationStatus := projectMigrationStatuses.GetProjectMigrationStatus(entity.ID(c.ProjectId))
		m.logger.Info("GetProjectMigrationStatus successfully",
			zap.Bool("projectMigrationStatus", projectMigrationStatus.Value()))

		tidbClusterInfo := datasource.TidbClusterInfo{
			Vendor:    c.CloudProviderName,
			Region:    c.RegionName,
			TenantID:  entity.ID(c.TenantId),
			ProjectID: entity.ID(c.ProjectId),
			ClusterID: entity.ID(c.Id),
		}
		ctx, err = datasource.SetMetricDataAddressToContextWithTidbClusterInfo(ctx, projectMigrationStatus, tidbClusterInfo)
		if err != nil {
			m.logger.Error("failed to call datasource.SetMetricDataAddressToContextWithTidbClusterInfo, skip this tidb cluster",
				zap.Bool("projectMigrationStatus", projectMigrationStatus.Value()),
				zap.String("tidb_cluster_id", id), zap.Error(err))
			continue
		} else {
			m.logger.Info("SetMetricDataAddressToContextWithTidbClusterInfo successfully",
				zap.Bool("projectMigrationStatus", projectMigrationStatus.Value()),
				zap.String("tidb_cluster_id", id), zap.Error(err))
		}

		// Get cluster's instances.
		thanosRes, err := central_thanos.QueryNodeMemoryUsedBytes(ctx, c.Id, metadata.M.TidbCloudNodeMemoryUsedBytes.Name())
		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.Bool("projectMigrationStatus", projectMigrationStatus.Value()),
				zap.String("tidb_cluster_id", id), zap.Error(err))
			continue
		} else {
			m.logger.Info("QueryNodeMemoryUsedBytes successfully",
				zap.Bool("projectMigrationStatus", projectMigrationStatus.Value()),
				zap.String("tidb_cluster_id", id), zap.Error(err),
				zap.Int("len_of_data", len(thanosRes.Data.Result)))
		}

		// For each instance, create memory/cpu/disk capacity metrics.
		for _, result := range thanosRes.Data.Result {
			labelSet := result.LabelSet
			valueTuple := result.Value
			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.Bool("projectMigrationStatus", projectMigrationStatus.Value()),
					zap.String("tidb_cluster_id", id), zap.Error(err))
				break
			}
			if labelSet.Component == tidbcomp.PD.String() {
				continue
			}
			resource := compMap[labelSet.Component]

			writeIntDatapoint(memSlice, resource.mem, now, projectName, c.Name, id, labelSet.Component, datasource.UnifyDBName(labelSet.Instance))
			writeDoubleDatapoint(cpuSlice, resource.cpu, now, projectName, c.Name, id, labelSet.Component, datasource.UnifyDBName(labelSet.Instance))
			writeIntDatapoint(diskSlice, resource.disk, now, projectName, c.Name, id, labelSet.Component, datasource.UnifyDBName(labelSet.Instance))
		}
	}

	return md, nil
}

// Add a single data point and label set to the metric family(pdata.Metric).
func writeIntDatapoint(m pmetric.Metric, value int64, t pcommon.Timestamp, projectName, clusterName, clusterID, component, instance string) {
	labels := pcommon.NewMap()
	labels.Insert(metadata.A.ProjectName, pcommon.NewValueString(projectName))
	labels.Insert(metadata.A.ClusterName, pcommon.NewValueString(clusterName))
	labels.Insert(metadata.A.ClusterID, pcommon.NewValueString(clusterID))
	labels.Insert(metadata.A.Component, pcommon.NewValueString(component))
	labels.Insert(metadata.A.Instance, pcommon.NewValueString(datasource.UnifyDBName(instance)))
	p := m.Gauge().DataPoints().AppendEmpty()
	p.SetTimestamp(t)
	p.SetIntVal(value)
	labels.CopyTo(p.Attributes())
}

func writeDoubleDatapoint(m pmetric.Metric, value float64, t pcommon.Timestamp, projectName, clusterName, clusterID, component, instance string) {
	labels := pcommon.NewMap()
	labels.Insert(metadata.A.ProjectName, pcommon.NewValueString(projectName))
	labels.Insert(metadata.A.ClusterName, pcommon.NewValueString(clusterName))
	labels.Insert(metadata.A.ClusterID, pcommon.NewValueString(clusterID))
	labels.Insert(metadata.A.Component, pcommon.NewValueString(component))
	labels.Insert(metadata.A.Instance, pcommon.NewValueString(datasource.UnifyDBName(instance)))
	p := m.Gauge().DataPoints().AppendEmpty()
	p.SetTimestamp(t)
	p.SetDoubleVal(value)
	labels.CopyTo(p.Attributes())
}

// parseIntegrations transforms integrations to a datadog api key map and a project id list.
func parseIntegrations(is []entity.Integration) (map[entity.ID]entity.Integration, []entity.ID) {
	m := make(map[entity.ID]entity.Integration, len(is))
	l := make([]entity.ID, 0, len(is))
	for _, i := range is {
		m[i.ProjectID] = i
		l = append(l, i.ProjectID)
	}
	return m, l
}