package clusters

import (
	"context"
	"fmt"
	"strings"

	"github.com/go-logr/logr"
	"github.com/prometheus/client_golang/prometheus"
	observabilitypingcapcomv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/structs"
	myclient "github.com/tidbcloud/observability/pkg/mysql/client"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

var (
	ClusterInfo = prometheus.NewDesc(
		prometheus.BuildFQName("observability", "cluster", "info"),
		"Information and status of o11y  cluster",
		[]string{"region", "cluster_id", "vendor", "tenantID", "projectID", "applicationID", "bizType", "vm_account_id", "vm_project_id", "mode", "status"},
		nil)

	ClusterTotal = prometheus.NewGaugeVec(prometheus.GaugeOpts{
		Namespace: "observability",
		Subsystem: "clusters",
		Name:      "total",
		Help:      "Number of clusters",
	}, []string{"biz_type", "vendor", "region", "from"})
)

type ClusterCollector struct {
	cli         client.Client
	rdsCli      myclient.Client
	log         logr.Logger
	serviceInfo structs.ServiceInfo
}

func NewClusterCollector(cli client.Client, rdsCli myclient.Client, log logr.Logger, serviceInfo structs.ServiceInfo) *ClusterCollector {
	return &ClusterCollector{
		cli:         cli,
		rdsCli:      rdsCli,
		log:         log,
		serviceInfo: serviceInfo,
	}
}
func (c *ClusterCollector) Describe(ch chan<- *prometheus.Desc) {
	ch <- ClusterInfo
}
func (c *ClusterCollector) Collect(ch chan<- prometheus.Metric) {
	ctx := context.Background()
	var clusterlist observabilitypingcapcomv1.ClusterList
	err := c.cli.List(ctx, &clusterlist)
	if err != nil {
		c.log.Error(err, "get clusters list failed!")
		return
	}
	totalmap := make(map[string]int32)
	for _, cluster := range clusterlist.Items {
		bizType, appID := cluster.Labels[constants.LabelBizType], cluster.Labels[constants.LabelApplicationID]
		// do not expose cluster info metrics for serverless child clusters.
		if !strings.Contains(bizType, "devtier") {
			ch <- prometheus.MustNewConstMetric(
				ClusterInfo,
				prometheus.GaugeValue,
				1.0,
				c.serviceInfo.LogicalRegionName,
				cluster.Name,
				c.serviceInfo.Vendor,
				cluster.Labels[constants.LabelTenantID],
				cluster.Labels[constants.LabelProjectID],
				appID,
				bizType,
				fmt.Sprintf("%d", cluster.Spec.VictoriaAccount.AccountID),
				fmt.Sprintf("%d", cluster.Spec.VictoriaAccount.ProjectID),
				cluster.Labels[constants.LabelShareMode],
				string(cluster.Status.Phase),
			)
		}
		totalmap[bizType]++
	}

	for bizType, count := range totalmap {
		ClusterTotal.WithLabelValues(
			bizType,
			c.serviceInfo.Vendor,
			c.serviceInfo.LogicalRegionName,
			"kubernetes",
		).Set(float64(count))
	}

	if c.serviceInfo.ControllerConfig.ServerlessConfig.EnableRDSClient {
		db := c.rdsCli.RawClient()

		// Query RDS to count clusters by biz_type
		type BizTypeCount struct {
			BizType string
			Count   int32
		}
		var bizTypeCounts []BizTypeCount

		err := db.WithContext(ctx).
			Model(&myclient.Cluster{}).
			Select("biz_type, COUNT(*) as count").
			Group("biz_type").
			Scan(&bizTypeCounts).Error

		if err != nil {
			c.log.Error(err, "failed to query cluster counts by biz_type from RDS")
		} else {
			// Merge RDS counts into totalmap
			for _, bc := range bizTypeCounts {
				ClusterTotal.WithLabelValues(
					bc.BizType,
					c.serviceInfo.Vendor,
					c.serviceInfo.LogicalRegionName,
					"rds",
				).Set(float64(bc.Count))
			}
		}

		// Query serverless parent clusters from RDS and expose cluster info metrics
		var serverlessParentClusters []myclient.Cluster
		err = db.WithContext(ctx).
			Preload("Spec").
			Preload("Status").
			Where("application_id = ?", constants.ApplicationIDServerless).
			Find(&serverlessParentClusters).Error

		if err != nil {
			c.log.Error(err, "failed to query serverless parent clusters from RDS")
		} else {
			for _, cluster := range serverlessParentClusters {
				ch <- prometheus.MustNewConstMetric(
					ClusterInfo,
					prometheus.GaugeValue,
					1.0,
					c.serviceInfo.LogicalRegionName,
					cluster.ClusterID,
					c.serviceInfo.Vendor,
					cluster.Tenant,
					cluster.ProjectID,
					cluster.ApplicationID,
					cluster.BizType,
					fmt.Sprintf("%d", cluster.Spec.VMAccountID),
					fmt.Sprintf("%d", cluster.Spec.VMProjectID),
					cluster.Spec.Labels[constants.LabelShareMode],
					cluster.Status.Phase,
				)
			}
		}
	}

}