package importinto

import (
	"context"
	"fmt"
	"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"
	"go.uber.org/zap"
)

const tenantLabel = "tenant"

var RecentTerminalStatusGauge = prometheus.NewDesc(
	prometheus.BuildFQName(metrics.DataflowNamespace, "import_into", "recent_terminal_status"),
	"Gauge set to 1 for each recently finished import task, labeled by import_id and terminal status",
	[]string{tenantLabel, metrics.ClusterIDLabel, "import_id", "status"},
	nil)

type StatusCollector struct {
	importIntoModel *pimport.ImportIntoModel
}

func shouldCountFailedImportIntoTask(im *pimport.ImportIntoTask) bool {
	if im == nil || im.Status != entity.ImportIntoStatusFailed {
		return false
	}
	return im.ImportResults.ControlPlaneFailed
}

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

func (c *StatusCollector) Describe(ch chan<- *prometheus.Desc) {
	ch <- RecentTerminalStatusGauge
}

func (c *StatusCollector) Collect(ch chan<- prometheus.Metric) {
	recentTime := time.Now().Add(-5 * time.Minute)
	imports, err := c.importIntoModel.ListRecentlyFinishedTerminalImportIntoTasks(context.Background(), recentTime)
	if err != nil {
		log.L().Error("Retrieve import job failed", zap.Error(err))
		return
	}
	c.collectFromImports(imports, ch)
}

func (c *StatusCollector) collectFromImports(imports []pimport.ImportIntoTask, ch chan<- prometheus.Metric) {
	for _, im := range imports {
		status := ""
		if im.Status == entity.ImportIntoStatusCompleted {
			status = string(entity.ImportIntoStatusCompleted)
		} else if shouldCountFailedImportIntoTask(&im) {
			status = string(entity.ImportIntoStatusFailed)
		}

		if status == "" {
			continue
		}

		ch <- prometheus.MustNewConstMetric(
			RecentTerminalStatusGauge,
			prometheus.GaugeValue,
			1,
			fmt.Sprintf("%d", im.Cluster.GetOrganizationId()),
			fmt.Sprintf("%d", im.ClusterID),
			fmt.Sprintf("%d", im.ID),
			status,
		)
	}
}