package importinto

import (
	"context"
	"strconv"
	"time"

	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus"
	pimport "github.com/tidbcloud/dataflow-service-ng/app/models/import_into"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	metrics "github.com/tidbcloud/dataflow-service-ng/internal/metrics"
	"github.com/tidbcloud/pkgv2/logid"
	"go.uber.org/zap"
)

// RegisterMetrics registers the metrics which are ONLY used in DbaaS server.
func RegisterMetrics(importIntoModel *pimport.ImportIntoModel) {
	prometheus.MustRegister(NewCollector(importIntoModel))
	prometheus.MustRegister(NewStatusCollector(importIntoModel))
}

var _ prometheus.Collector = &Collector{}

var (
	syncingImportIntoTaskSinceCreatedSeconds = prometheus.NewDesc(
		prometheus.BuildFQName(metrics.DataflowNamespace, "import", "syncing_import_into_task_since_created_seconds"),
		"How long has the import been created still syncing",
		[]string{tenantLabel, metrics.ClusterIDLabel, "import_id", "status"},
		nil)
	importIntoTaskCreatingDurationSeconds = prometheus.NewDesc(
		prometheus.BuildFQName(metrics.DataflowNamespace, "import_into_task", "creating_duration_seconds"),
		"How long an import-into task has stayed in preparing or scaling",
		[]string{tenantLabel, metrics.ClusterIDLabel, "import_id", "status"},
		nil)
)

type Collector struct {
	importIntoModel *pimport.ImportIntoModel
}

func NewCollector(importIntoModel *pimport.ImportIntoModel) prometheus.Collector {
	return &Collector{
		importIntoModel: importIntoModel,
	}
}

func (c *Collector) Describe(ch chan<- *prometheus.Desc) {
	ch <- syncingImportIntoTaskSinceCreatedSeconds
	ch <- importIntoTaskCreatingDurationSeconds
}

func (c *Collector) Collect(ch chan<- prometheus.Metric) {
	logger := log.L().With(zap.String("collector", "import"), zap.String("runID", logid.GenLogid()))
	if err := c.collect(ch, logger); err != nil {
		logger.Error("failed to collect metrics", zap.Error(err))
	}
}

func (c *Collector) collect(ch chan<- prometheus.Metric, logger *zap.Logger) (err error) {
	start := time.Now()
	defer func() {
		if err != nil {
			logger.Error("failed to collect metrics", zap.Error(err))
		} else {
			logger.Info("imports metrics collect time", zap.Int64("time", time.Since(start).Milliseconds()))
		}
	}()
	importIntoTasks, err := c.importIntoModel.ListIncompletedImportIntoTasks(context.Background())
	if err != nil {
		return err
	}

	for _, imp := range importIntoTasks {
		dur := time.Since(imp.CreatedAt).Seconds()

		ch <- prometheus.MustNewConstMetric(
			syncingImportIntoTaskSinceCreatedSeconds,
			prometheus.GaugeValue,
			dur,
			strconv.FormatUint(imp.Cluster.GetOrganizationId(), 10),
			strconv.FormatUint(imp.ClusterID.Uint64(), 10),
			strconv.FormatUint(imp.ID.Uint64(), 10),
			string(imp.Status),
		)

		if imp.Status == entity.ImportIntoStatusPreparing || imp.Status == entity.ImportIntoStatusScalingUp {
			ch <- prometheus.MustNewConstMetric(
				importIntoTaskCreatingDurationSeconds,
				prometheus.GaugeValue,
				dur,
				strconv.FormatUint(imp.Cluster.GetOrganizationId(), 10),
				strconv.FormatUint(imp.ClusterID.Uint64(), 10),
				strconv.FormatUint(imp.ID.Uint64(), 10),
				string(imp.Status),
			)
		}
	}
	return nil
}