package query

import (
	"bytes"
	"context"
	"crypto/tls"
	"sort"

	"github.com/pingcap/dbaas/central/config"
	"github.com/pingcap/dbaas/central/db"
	"github.com/pingcap/dbaas/central/db/dbutil"
	"github.com/pingcap/dbaas/central/db/entity"
	infraapi "github.com/pingcap/dbaas/central/infra-api"
	"github.com/pingcap/dbaas/central/metrics/prom"
	"github.com/pingcap/log"
	"github.com/tidbcloud/metrics/internal/datasource"
	"github.com/tidbcloud/pkgv2/infra/aws"
	panicIf "github.com/tidbcloud/pkgv2/panicif"
	"go.uber.org/zap"
	"gorm.io/gorm"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/utils/pointer"
)

type Manager struct {
	api       *infraapi.API
	proxyCfg  config.ProxyConfig
	tlsConfig *tls.Config
}

func NewManager() *Manager {
	proxyConfig := config.Global().ProxyConfig

	tlsConfig, err := aws.ProxyTlsConfig(config.Global().AwsConfig.PCAArn)
	panicIf.Error(err)

	// Gardener cluster k8s client.
	var cfg *rest.Config
	kubeConfig := config.Global().GardenerConfig.KubeConfigPath
	if kubeConfig == "" {
		cfg, err = rest.InClusterConfig()
	} else {
		cfg, err = clientcmd.BuildConfigFromFlags("", kubeConfig)
	}
	panicIf.Error(err)
	// Initialize the k8s/gardener clients.
	baseKubeCli, err := kubernetes.NewForConfig(cfg)
	panicIf.Error(err)
	infraAPI := infraapi.InitInfraAPI(baseKubeCli)

	return &Manager{
		api:       infraAPI,
		proxyCfg:  proxyConfig,
		tlsConfig: tlsConfig,
	}
}

func (m *Manager) QueryMetrics(projectID uint64, cluster *entity.Cluster, cloud entity.Cloud, useClusterMetrics bool) ([]byte, error) {
	clusterID := cluster.ID.Uint64()

	client, err := prom.NewBasePromClient(projectID, clusterID, m.api, m.proxyCfg, m.tlsConfig)
	if err != nil {
		return nil, err
	}

	type result struct {
		value prom.Metric
		err   error
		index int
	}

	var allMetricQueries []datasource.MetricQuerierFunc
	if useClusterMetrics {
		allMetricQueries = registryClusterMetrics(client, clusterID, cluster.Name)
	} else {
		allMetricQueries = registryMetrics(client, clusterID)
	}

	ctx := context.Background()
	resultsCh := make(chan result, len(allMetricQueries))
	for idx, query := range allMetricQueries {
		go func(idx int, query datasource.MetricQuerierFunc) {
			m, err := query(ctx)
			resultsCh <- result{
				value: m,
				err:   err,
				index: idx,
			}
		}(idx, query)
	}
	results := make([]result, 0, len(allMetricQueries))
	for range allMetricQueries {
		r := <-resultsCh
		if r.err != nil {
			log.Warn("GetMeasurements API failed to query metrics",
				zap.Uint64("clusterID", clusterID),
				zap.Error(r.err))
		}
		results = append(results, r)
	}
	sort.Slice(results, func(i, j int) bool {
		return results[i].index < results[j].index
	})
	outputData := bytes.NewBuffer(make([]byte, 0, 10*1024))
	for _, r := range results {
		if r.err == nil {
			_ = r.value.EncodePrometheusFormat(outputData)
		}
	}

	return outputData.Bytes(), nil
}

func (m *Manager) QueryProjectMetrics(projectID uint64) ([]byte, error) {
	var (
		err      error
		clusters []entity.Cluster
	)
	err = db.Transaction(func(tx *gorm.DB) error {
		id := entity.ID(projectID)

		clusters, err = dbutil.FindClusters(tx, dbutil.ClusterFilter{
			ProjectID: &id,
			Deleted:   pointer.BoolPtr(false),
			Cleared:   pointer.BoolPtr(false),
			Creating:  pointer.BoolPtr(false),
		})

		return err
	})
	if err != nil {
		log.Warn("failed to query clusters from projectID", zap.Uint64("projectID", projectID), zap.Error(err))
		return nil, err
	}

	type clusterMetricsResp struct {
		clusterID uint64
		data      []byte
		err       error
	}
	clusterMetricsCh := make(chan clusterMetricsResp, len(clusters))
	for _, cluster := range clusters {
		go func(cluster entity.Cluster) {
			data, err := m.QueryMetrics(projectID, &cluster, cluster.Provider.Name, true)
			clusterMetricsCh <- clusterMetricsResp{
				clusterID: cluster.ID.Uint64(),
				data:      data,
				err:       err,
			}
		}(cluster)
	}

	outputData := bytes.NewBuffer(make([]byte, 0, 100*1024))
	for i := 0; i < len(clusters); i++ {
		r := <-clusterMetricsCh
		if r.err != nil {
			log.Warn("failed to query cluster metrics", zap.Uint64("clusterID", r.clusterID), zap.Error(r.err))
			continue
		}
		outputData.Write(r.data)
	}

	return outputData.Bytes(), nil
}